datacollector.java

电信数据采集系统,对unix下的一个日志文件进行采集,解析,并进行匹配

package dms.business.collectProcess;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.MappedByteBuffer;
import java.util.Vector;

import dms.business.sendProcess.CollectionListener;
import dms.business.sendProcess.LogDealer;
import dms.data.LogRecord;
import dms.data.MatchedRecord;
import dms.exception.HistoryFileException;
import dms.exception.HistoryLocationFileException;
import dms.exception.NetworkException;
import dms.exception.SourceFileException;
import dms.exception.StoreFileException;
import dms.util.DataReader;
import dms.util.FileUtil;

/**
 * 数据采集器
 */
public class DataCollector {
	/**
	 * 采集主方法
	 * @throws SourceFileException 
	 * @throws HistoryLocationFileException 
	 * @throws HistoryFileException 
	 * @throws StoreFileException 
	 * @throws NetworkException 
	 */
	public void collect() throws SourceFileException, 
												HistoryLocationFileException, 
												HistoryFileException, 
												StoreFileException, 
												NetworkException  {
		Vector<LogRecord> logins = new Vector<LogRecord>();
		Vector<LogRecord> logouts = new Vector<LogRecord>();
//		step1 读源文件，并映射到本地缓存
		String sourceFile = FileUtil.getSourceFile();
		DataReader dataReader = new DataReader();
		MappedByteBuffer buffer;
		try {
			buffer = dataReader.mappingLogBuffer(sourceFile);
		} catch(FileNotFoundException e){
			throw new SourceFileException();
		}catch (IOException e) {
			throw new SourceFileException();
		}
//		step1.1 判断historyFile是否存在
		String historyFileName = FileUtil.getHistoryFile();//上次匹配未成功的记录
		if(new File(historyFileName).exists()){
			Object obj;
			try {
				obj = FileUtil.active(historyFileName);
			} catch (IOException e) {
				throw new HistoryFileException();
			} catch (ClassNotFoundException e) {
				throw new HistoryFileException();
			}
			Vector<LogRecord> historyLogins = null;
			if(obj instanceof Vector){
				historyLogins = (Vector<LogRecord>) obj;
				System.out.println("上次匹配失败的记录"+ historyLogins.size()+"条");
				copyHistoryLogins( historyLogins,logins);//把上次没匹配上的记录拷贝到本次Vector，一起匹配		
			}
		}
//		step2 解析文件，过滤记录
		DataParser dataParser = new DataParser();
		try {
			dataParser.parseLogBuffer(buffer, logouts, logins);
		} catch (IOException e) {
			throw new HistoryLocationFileException();
		} catch (ClassNotFoundException e) {
			throw new HistoryLocationFileException();
		}
//		step3 匹配记录
		DataMatcher dataMatcher = new DataMatcher();
		Vector<MatchedRecord> matchedRecords;
		try {
			matchedRecords = dataMatcher.match(logouts, logins);
		} catch (IOException e) {
			throw new HistoryFileException();
		}
//		step4 发送匹配好的记录
		CollectionListener logdealer = new LogDealer();
		try {
			logdealer.deal(matchedRecords);
		} catch (IOException e) {
			throw new NetworkException();
		} catch (ClassNotFoundException e) {
			throw new StoreFileException();
		}
	}

	/**
	 * 把上次未匹配上的historyLogins，装入这次logins匹配
	 * @param historyLogins 上次未匹配上的登录数据
	 * @param logins 这次登录数据
	 */
	private void copyHistoryLogins(Vector<LogRecord> historyLogins, Vector<LogRecord> logins) {
		if(!historyLogins.isEmpty()){
			for(LogRecord login:historyLogins){
				logins.addElement(login);
			}
		}
	}

}