autochangemessage.java

自动导入文件到数据库,用于那些无法实时入库,定点时间入库的需求

package com.onet.autobill.changemessage;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.regex.Pattern;

import javax.sql.DataSource;

import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;

import com.onet.autobill.changemessage.db.UpdateOfChangeMessage;

//import com.onet.util.tools.Locker;

public class AutoChangeMessage {

	private static Logger log = Logger.getLogger(AutoChangeMessage.class);

	private ChangeMessageConfig changeMessageConfig = null;

	private List<String> fileNamesList = new ArrayList<String>();
	
	private List<String> list = new LinkedList<String>();
	
	private List<File> fileList = new LinkedList<File>();
	
	private List<OutputStreamWriter> FileOutputStreamList = new LinkedList<OutputStreamWriter>();
	
	private UpdateOfChangeMessage updateOfChangeMessage = new UpdateOfChangeMessage();
	
	private String configFile = "config/applicationContext.xml";

	private String nowDate;
	
	private DataSource dbSource;
	
	private String fileNamePart = "";
	
	private String fileAbsolutePath = "";
	
	private String TABLENAME_SQL;
	
	private String[] fileNames = null;

	private boolean readConfig() {
		try {

			ApplicationContext appContext = new FileSystemXmlApplicationContext(
					configFile);
			changeMessageConfig = (ChangeMessageConfig) appContext
					.getBean("changeMessageConfig");
			
			dbSource = changeMessageConfig.getDbSource();
			fileNamePart = changeMessageConfig.getFileNamePart();
			fileAbsolutePath = changeMessageConfig.getFileAbsolutePath();
			TABLENAME_SQL = changeMessageConfig.getTABLENAME_SQL();
			
			File file = new File(fileAbsolutePath);
			file.mkdir();
			
		} catch (Exception e) {
			log.error("读取配置文件错误！", e);
			return false;
		}

		return true;
	}

	public boolean combineStandardFile(String sourceFilePath) {
		
		try {
			String gateWay ="";
			String channel = "";
			String textfileName = "";
			File parentFile = new File(sourceFilePath);
			findFilenames(parentFile);
			if(fileNamesList.size()==0){
				log.warn("没有符合条件的文件");
				return false;
			}else{
				log.info(sourceFilePath+"下有"+fileNamesList.size()+"个符合条件的文件");
			}
			int index = 0;
			Iterator<String> it = fileNamesList.iterator();
			while (it.hasNext()) {
				String fileName = it.next();
				File changeFile = new File(fileName);
				InputStream fin = new FileInputStream(changeFile);
				BufferedReader rndFile = new BufferedReader(
						new InputStreamReader(fin));
				
				for (String str = null; (str = rndFile.readLine()) != null;) {
					log.info(str);
					
					int point = fileName.indexOf(".");
					String date = fileName.substring(point-8, point);
					gateWay = str.substring(str.length()-2);
					str = str +"\t"+"0" +"\r\n";	//此处有个0，表示没有处理，到后面的存储过程会更新
					textfileName = gateWay+fileNamePart;
					log.info("文件名称："+textfileName+"_"+date+".txt");
					
					if(list.size()==0){
						list.add(gateWay);
						fileList.add(new File(textfileName+"_"+date+".txt"));
						FileOutputStreamList.add(new OutputStreamWriter(new FileOutputStream(fileAbsolutePath+"\\"+fileList.get(index),true)));
						log.info(textfileName+"不存在，集合为空");
					}			
					
					for(int i =0;i<list.size();i++){
						channel = list.get(i);
						if(channel.equals(gateWay)){
							index = i;
							log.info(textfileName+"已经存在，集合不为空");
							break;
						}else{
							if(++i ==list.size() ){
								index = i;
								list.add(gateWay);
								fileList.add( new File(textfileName+"_"+date+".txt"));
								FileOutputStreamList.add(new OutputStreamWriter(new FileOutputStream(fileAbsolutePath+"\\"+fileList.get(index),true)));
								log.info(textfileName+"不存在，集合不为空");
								break;
							}else{
								--i;
								continue;
							}
						}

					}
					log.info(index);
					FileOutputStreamList.get(index).write(str);
					FileOutputStreamList.get(index).flush();

				}
				//rndFile.close();
				rndFile = null;
				fin = null;
				changeFile = null;
			}
			fileNamesList.clear();
			parentFile = null;
		} catch (Exception e) {
			log.error("合并文件错误！", e);
			return false;
		}

		return true;
	}

	/**
	 * 递归查找给定目录，找出符合条件的文件名数组
	 * 
	 * @param parentFile
	 *            查找路径文件
	 */
	private void findFilenames(File parentFile) {

		// 检查当前给定的文件是否存在并且是目录
		Date date = new Date();
		SimpleDateFormat sp = new SimpleDateFormat("yyyyMMdd");
		nowDate = sp.format(date);
		if (parentFile != null && parentFile.exists()
				&& parentFile.isDirectory()) {
			// 得到当前目录下所有文件、文件夹名称
			String[] filenameArr = parentFile.list();
			if (filenameArr != null) {
				// 构造成File对象，检查它是否是子目录
				for (int i = 0; i < filenameArr.length; i++) {
					String fileName = parentFile.getPath() + "\\"
							+ filenameArr[i];
					File currentFile = new File(fileName);
					// 如果是目录，则递归调用
					if (currentFile.isDirectory()) {
						this.findFilenames(currentFile);
					} else {
						try {
							if (Pattern.matches(changeMessageConfig
									.getFileFilter(), filenameArr[i])) {
								String fileDate = filenameArr[i].split("_")[2].substring(0, 8);
								if (fileDate != null
										&& fileDate.compareTo(nowDate) == 0) {
									fileNamesList.add(fileName);
								}
							}

						} catch (Exception e) {
							log.error(filenameArr[i], e);
						}

					}

				}
			}
		}

	}
	
	public boolean bulkIntoDatabase(String filePart) {
		
		String tableName ="";
		int connectFlg = 1;
		File findFile = new File(filePart);
		
		if (findFile != null && findFile.exists()) {
			fileNames = findFile.list();	
		}

		for(int i =0;i<fileNames.length;i++){
		    try {
					//获取表名，根据文件名称
					//tableName = getTableName(fileNames[i]);
		    		tableName = updateOfChangeMessage.getTableName(fileNames[i]);
					
					if(tableName ==null || tableName.equals(""))
							return false;
					//入库
					String cmd = "bcp " + tableName + " in "
							+ fileAbsolutePath+"//"+fileNames[i] + " -c -S"
							+ changeMessageConfig.getDbHost() + " -U"
							+ changeMessageConfig.getDbUser() + " -P"
							+ changeMessageConfig.getDbPassword();
					log.info(cmd);
					connectFlg = ConnectHost.executeCommand(cmd);
					if(connectFlg==1)
						return false;
			} catch (Exception e) {
				log.error("导入数据库错误！", e);
				return false;
			}
		}


		return true;
	}
	
	
	public boolean deleteTempFile() {
		for(int i=0;i<fileNames.length;i++){
			try {
				File file = new File(fileAbsolutePath+"\\"+fileNames[i]);
				file.delete();
				File filePath = new  File(fileAbsolutePath);
				filePath.delete();
	
			} catch (Exception e) {
				log.error("删除文件出错", e);
				return false;
			}
		}

		return true;
	}

	public boolean updateConfigFile(String configFileName) {
		FileInputStream fileInStream = null;
		FileOutputStream fileOutStream = null;
		try {
			changeMessageConfig.setLastSyncTime(nowDate);
			Properties properties = new Properties();
			fileInStream = new FileInputStream(configFileName);
			properties.load(fileInStream);
			properties.setProperty("sysytem.lastsynctime", nowDate);
			fileOutStream = new FileOutputStream(configFileName);
			properties.store(fileOutStream, null);

		} catch (Exception e) {
			log.error("更新配置文件出错", e);

			return false;
		} finally {
			if (fileInStream != null) {
				try {
					fileInStream.close();

				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (fileOutStream != null) {
				try {
					fileOutStream.close();

				} catch (IOException e) {
					e.printStackTrace();
				}
			}

		}

		return true;
	}

	public void work() {

//		// 确保当前目录下只有一个进程在运行
//		Locker locker = new Locker();
//		if (!locker.lock()) {
//			System.err.print("程序已经启动了!");
//			System.exit(-1);
//		}

		
		// 读取配置文件
		log.info("读取配置文件");
		if (false == readConfig()) {
			return;
		}
		log.info("读取配置文件成功！");
		
		updateOfChangeMessage.setDbSource(dbSource);
		updateOfChangeMessage.setTABLENAME_SQL(TABLENAME_SQL);

		// 合并文件
		Iterator<String> it = changeMessageConfig.getFilePath().iterator();		
		while (it.hasNext()) {
			combineStandardFile(it.next());
		}
		log.info("合并文件成功！");
		
		//一定要注意这里，必须把文件流关掉，不然会报 ：无法打开 BCP 主数据文件  错误
		for(int i =0 ;i<FileOutputStreamList.size();i++){
			try {
				FileOutputStreamList.get(i).close();
			} catch (IOException e) {
				log.debug("关闭文件失败!");
			}
		}
		FileOutputStreamList = null;
		fileNamesList = null;
		list = null;
		fileList = null;
		// 导入到数据库中
		if (false == bulkIntoDatabase(fileAbsolutePath)) {
			return;
		}
		log.info("导入数据库成功！");
		
		// 删除临时文件
		if (false == deleteTempFile()) {
			return;
		}
		log.info("删除临时文件成功！");
		
		changeMessageConfig = null;
		dbSource = null;
		fileNames = null;
		
		updateOfChangeMessage.changeMessageUpdate();
	}

//	public String getTableName(String dateStr) throws SQLException {
//		
//		// 返回的表的名称
//		String tableName = null;
//		Connection con = null;
//		CallableStatement CS = null;
//		
//		try{
//			con = dbSource.getConnection();
//			if (con == null) {
//				log.error("数据库连接失败!");
//				return "";
//			}
//			CS = con.prepareCall(TABLENAME_SQL);
//			CS.clearParameters();
//			// 注册返回值
//			CS.registerOutParameter(2, Types.VARCHAR);
//			
//			CS.setString(1, dateStr);
//			
//			CS.execute();
//			
//			tableName = CS.getString(2);
//			
//		}catch(SQLException ex){
//			log.warn("取表名失败!", ex);
//			return "";
//		} finally {
//			try {
//				if (con != null)
//					con.close();
//			} catch (Exception e) {
//				log.warn("关闭连接失败！", e);
//			}
//		}
//		CS.close();
//
//		return tableName;
//	}
	
//	public static void main(String[] parameters) {
//		AutoChangeMessage as = new AutoChangeMessage();
//		as.work();
//	}

	public void setChangeMessageConfig(ChangeMessageConfig changeMessageConfig) {
		this.changeMessageConfig = changeMessageConfig;
	}

	public String getFileAbsolutePath() {
		return fileAbsolutePath;
	}

	public void setFileAbsolutePath(String filePath) {
		fileAbsolutePath = filePath;
	}

	public void setDbSource(DataSource dbSource) {
		this.dbSource = dbSource;
	}

	public void setTABLENAME_SQL(String tablename_sql) {
		TABLENAME_SQL = tablename_sql;
	}


	public void setFileNames(String[] fileNames) {
		this.fileNames = fileNames;
	}

	public void setNowDate(String nowDate) {
		this.nowDate = nowDate;
	}

	public String getFileNamePart() {
		return fileNamePart;
	}

	public void setFileNamePart(String fileNamePart) {
		this.fileNamePart = fileNamePart;
	}

}