hbase 报文 处理 逻辑

 

 

报文格式：

 

每隔一个小时 出现一个文件类型 报文 ，所以 我们的处理思路是，一个小时做一次处理。

 

import java.io.FileInputStream;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 

import com.cmcc.aoi.util.OsUtil;

public class HbaseStarter {

	public static void main(String[] args) throws  Exception {
		Properties properties=new Properties();
		String config="";
		if(!OsUtil.isLinux())
			config= "D:/work/util/aoi-hbase/trunk/src/main/resources/config.properties";
		else	
			config = "/home/aoi/aoi-hbase/conf/config.properties"; 
		FileInputStream fis = new FileInputStream(config);
		properties.load(fis);
		fis.close(); 
		
		String hbaseTable = properties.getProperty("com.cmcc.aoi.ua.hbaseTable");
		String hbaseFamily = properties.getProperty("com.cmcc.aoi.ua.hbaseFamily"); 
		String sourceFilePath=properties.getProperty("com.cmcc.aoi.ua.sourceFilePath");
		String archivelogsPath=properties.getProperty("com.cmcc.aoi.ua.archivelogsPath");
		boolean isDebug= Integer.parseInt( properties.getProperty("com.cmcc.aoi.ua.isDebug")) == 0 ? false : true;
		String sourceFileName = properties.getProperty("com.cmcc.aoi.ua.sourceFileName"); 
		 
		String[] hbaseTableName=hbaseTable.split(",");  // table
		String[] hbaseFamilyName=hbaseFamily.split("&");// family  
		String[] sourceFileNameArr=sourceFileName.split(","); 
		 
		
		ScheduledExecutorService service = Executors.newScheduledThreadPool(2);  
		service.scheduleAtFixedRate(new DeviceReadThread (sourceFileNameArr[0],hbaseTableName[0],hbaseFamilyName[0].split(","),sourceFilePath,archivelogsPath,isDebug) 
			               ,0, 1,TimeUnit.HOURS); 
		service.scheduleAtFixedRate(new AppReadThread (sourceFileNameArr[1],hbaseTableName[1],hbaseFamilyName[1].split(","),sourceFilePath,archivelogsPath,isDebug) 
                         ,0, 1,TimeUnit.HOURS); 
	}
}

 

 

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader; 
import java.io.IOException;
import java.util.Date;  
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; 
import Model.Device; 
import com.alibaba.fastjson.JSON; 

public class DeviceReadThread extends BaseRunnabler {

	static Logger logger = LoggerFactory.getLogger(DeviceReadThread.class);
	
	public DeviceReadThread(String sourceFileName, String hbaseTable,
			String[] hbaseFamily, String sourceFilePath,
			String archivelogsPath, boolean isDebug) {
		super(sourceFileName, hbaseTable, hbaseFamily, sourceFilePath, archivelogsPath,
				isDebug); 
	}
	 
	public void processFile(IOperator hu) {
		FileReader logReader = null;
		BufferedReader logBufferedReader = null;
		try { 
			File logFile = new File(sourceFilePath+sourceFileName);
			logReader = new FileReader(logFile);
			logBufferedReader = new BufferedReader(logReader);
			String temp = logBufferedReader.readLine();
			//logger.error(" temp is  " + temp );
			while ( temp  != null) {
				Device device = JSON.parseObject(temp, Device.class); 
				//logger.error(" device is null ? " + ( device == null ) );
				
				String[][] s = new String[][] {
						{ device.getLid(), hbaseFamily[0], "lid" , device.getLid() } ,
						{ device.getLid(), hbaseFamily[1], "date", (new Date()).toString() }, 
						{ device.getLid(), hbaseFamily[2], "os", device.getOs() },
						{ device.getLid(), hbaseFamily[2], "osv", device.getOsv()} };
				hu.writeMultRow(hbaseTable, s);
				logger.info(" hbase util end "   );
				temp = logBufferedReader.readLine();
			}
		} catch (Exception e) {
			logger.error(" DeviceReadThread error "   );
			e.printStackTrace();
		} finally { 
			try {
				logBufferedReader.close();
			} catch (IOException e) { 
				e.printStackTrace();
			}
			try {
				logReader.close();
			} catch (IOException e) { 
				e.printStackTrace();
			}
		}
	} 

}

 

import java.io.File; 
import java.util.Arrays; 

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 
import com.cmcc.aoi.util.FileUtil;

public abstract class BaseRunnabler implements Runnable{
	protected static Logger logger = LoggerFactory.getLogger(BaseRunnabler.class);
	
	String sourceFileName=""; // 读取文件路径  
	String hbaseTable="";  // hbase  表名
	String [] hbaseFamily=null;    // 行列簇名 
	String sourceFilePath ;
	String archivelogsPath ;
	boolean isDebug; 
		
	
	public BaseRunnabler(String sourceFileName,String hbaseTable,String [] hbaseFamily  ,String sourceFilePath,String archivelogsPath,boolean isDebug ){
		this.hbaseTable=hbaseTable;
		this.hbaseFamily = hbaseFamily; 
		this.sourceFileName=sourceFileName;  
		this.sourceFilePath = sourceFilePath;
		this.archivelogsPath = archivelogsPath;
		this.isDebug = isDebug; 
	}	
	
	@Override
	public void run() {
		try{
		IOperator hu = new HbaseUtil( hbaseTable,hbaseFamily);
    	hu.createTable(hbaseTable,hbaseFamily );  
    	
    	File file=new File(sourceFilePath);
		 File[] tempFileList = file.listFiles();
		 Arrays.sort(tempFileList);
		 for (File tempFile: tempFileList) {
			 if (tempFile.isFile() && tempFile.getName().contains(sourceFileName +".") ) {
				try{
					try{
						processFile(hu);
					}catch (Exception e) { 
						logger.error("readfile error ,must continue to protect  to read  other file ");
						continue;
					}
				
				removeFile(tempFile);
				
				}catch (Exception e2) { 
					logger.error(" one file has an error ,other  file must continue to do this task ");
				}
			}
		 }
    	
    	 
		}catch (Exception e) {
			e.printStackTrace();
		}

	}
	
	public abstract void processFile(IOperator hu) throws Exception; 	
	
	private void removeFile(File file) {
		if (isDebug) {
			 File path = new File(archivelogsPath);
			 if (!path.exists()) {
				 path.mkdirs();
			 }
			 FileUtil.moveFile(file, new File(archivelogsPath,file.getName()));
			 logger.info("remove file :" + file.getName());
		 }else{
			 file.delete();
			 logger.info("delete file :" + file.getName());
		 }
	}
}

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

捐助开发者

在兴趣的驱动下,写一个免费的东西，有欣喜，也还有汗水，希望你喜欢我的作品，同时也能支持一下。 当然，有钱捧个钱场（右上角的爱心标志，支持支付宝和PayPal捐助），没钱捧个人场，谢谢各位。

 
 
 谢谢您的赞助，我会做的更好！