dataparser.java

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

package dms.business.collectProcess;

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

import dms.data.LogRecord;
import dms.util.FileUtil;

public class DataParser {
	private String historyLocationFile;
	private String labIP;
	/**
	 * 解析缓冲的内存结构，
	 * @param logBuffer 要解析的本地内存缓冲
	 * @param logout 返回解析后的登出数据
	 * @param login 返回解析后的登录数据
	 * @throws ClassNotFoundException  historyLocationFile异常
	 * @throws IOException  historyLocationFile异常
	 */
	public void parseLogBuffer(MappedByteBuffer logBuffer,
												Vector<LogRecord> logouts,
												Vector<LogRecord> logins) throws IOException, ClassNotFoundException   {
		init();
		long bufferSize = logBuffer.capacity();//缓冲区容量
		int size = (int) (bufferSize/372);//日志记录条数(每条记录372字节)
		System.out.println("size:"+size);
		int historyLocation = 0;//上次采集结束位置
		if(new File(historyLocationFile).exists()){
			historyLocation =(Integer) FileUtil.active(historyLocationFile);
			System.out.println("historyLocation:"+historyLocation);
		}
		int count = 0;
		int number=500;
		int j=historyLocation;
		int longinnumber = 0;
		for(int i = historyLocation ; i < size ; i++){
			if(count++ == number){//如果本次采集超过100条，则结束本次采集，防止源文件太大，导致内存溢出。
				break;
			}
			
			logBuffer.position(i*372);//定位在每条记录的开头位置
			byte[] loginName = new byte[32];//000-031字节代表用户登录名
			logBuffer.get(loginName);
			String userLoginName = new String(loginName).trim();//用户登录名
			logBuffer.get(new byte[36]);//跳过无用的字段
			int processID = logBuffer.getInt();//进程ID
			short type = logBuffer.getShort();//该条记录的类型：7代表登录，8代表登出
			logBuffer.get(new byte[6]);//跳过无用的字段
			long visitTime = logBuffer.getInt()*1000L;//在日志文件中生成该条记录的时间(毫秒)
			visitTime+=logBuffer.getInt();//累加毫秒  			
			logBuffer.get(new byte[26]);//跳过无用的字段
			byte[] remote = new byte[257];//记录ip
			logBuffer.get(remote);
			String userIP = new String(remote).trim();
			
			if(!userLoginName.startsWith(".") && (type == 7 || type == 8) && 
					(!userLoginName.equals("root"))){//过滤掉用户名以.开头的记录
				LogRecord logRecord = new LogRecord();
				logRecord.setUserIp(userIP);
				logRecord.setPid(processID);
				logRecord.setUserName(userLoginName);
				logRecord.setVisitTime(visitTime);
				logRecord.setLabIp(labIP);
				
//				Date date = new Date(visitTime);
//				System.out.println("userName: "+userLoginName+"\t pid: "+processID+"\t type: "+type+"\t logTime: "+date+"\t ip:"+userIP);
				if(type == 7){//登录记录放入login Vector
					logins.addElement(logRecord);
					longinnumber++;
				}
				if(type == 8){//登出记录放入logout Vector
					logouts.addElement(logRecord);
				}
			}
			historyLocation = i+1;
		}
		
		FileUtil.passivate(historyLocationFile, historyLocation);//把本次采集结束位置写入文件,下次采集直接定位到这个位置，开始采集
		if(size-j > number){
			System.out.println("本次一共解析"+number+"条"+"。登录："+longinnumber+"条"+"，登出："+logouts.size()+"条");
		}else{
			System.out.println("本次一共解析"+(size-j)+"条"+"。登录："+longinnumber+"条"+"，登出："+logouts.size()+"条");
		}		
	}
	
	private void init(){
		historyLocationFile = FileUtil.getHistoryLocationFile();
		labIP = FileUtil.getLabIP();
	}
}