logdealer.java

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

package dms.business.sendProcess;

import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Vector;

import dms.data.MatchedRecord;
import dms.util.FileUtil;

public class LogDealer implements CollectionListener {

	private String serverIp;
	private int serverPort;
	private String storeFile;
	private Socket socket;
	private DataInputStream inputStream;
	private ObjectOutputStream outputStream;
	
	public void deal(Vector<MatchedRecord> matches) throws IOException, ClassNotFoundException {
		init();
		initSocket();
		//如果上次发送失败，则将保存记录的文件中的数据读入本次matches
		if(new File(storeFile).exists()){
			Vector<MatchedRecord> tempRecords = null;
			Object obj = FileUtil.active(storeFile);
			if(obj instanceof Vector){
				tempRecords = (Vector<MatchedRecord>) obj;
				matches.addAll(tempRecords);
			}
		}
		send(outputStream,matches);
		close();
	}	

	/**
	 * 发送数据
	 * @param out 发送数据的网络流
	 * @param data 要发送的数据
	 * @throws IOException 
	 * @throws Exception 
	 */
	private void send(ObjectOutputStream out,Vector<MatchedRecord> data) throws IOException  {
		int successNumber = 0;
		for(MatchedRecord matched:data){
			out.writeObject(matched);
			out.flush();
			if(recieve(inputStream) == 0){
				System.out.println("发送成功！");
				successNumber++;
			}else{
				System.out.println("服务器端接收数据出错！");
				FileUtil.passivate(storeFile, data);
			}
		}
		System.out.println("成功发送:"+successNumber+"条记录");
		out.writeObject(null);//客户端发送完所有数据后发送一个null，以便服务器端退出接收循环
		out.flush();
	}
	
	/**
	 * 若client发送data成功，server会发送成功标志给予响应
	 * @param in网络接收流
	 * @return server是否处理成功标记，0：表示处理成功，否则表示处理失败
	 * @throws IOException 
	 */
	private int recieve(DataInputStream in) throws IOException{
		int flag = -1;
		flag = in.readInt();
		return flag;
	}

	/**
	 * 主要初始化服务器参数
	 * 配置参数从属性文件获取
	 */
	private void init(){
		serverIp = FileUtil.getServerIP();
		serverPort = Integer.parseInt(FileUtil.getServerPort());
		storeFile = FileUtil.getStoreFile();
	}
	
	/**
	 * 初始化网络连接
	 *
	 */
	private void initSocket() throws IOException{
		socket = new Socket(serverIp,serverPort);
		outputStream = new ObjectOutputStream(socket.getOutputStream());
		inputStream = new DataInputStream(socket.getInputStream());
	}
	
	/**
	 * 关闭流，关闭socket
	 * @throws IOException
	 */
	private void close() throws IOException {
		outputStream.close();
		inputStream.close();
		socket.close();
	}
}