import net.sf.json.JSONObject; import net.sf.json.JSONArray; import java.io.*; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.regex.Pattern; import java.util.regex.Matcher; public class MainDataAnalysis { /** * 程序主入口-计算核心的串行实现 * * return jsonOfAnswer JSONObject */ public static JSONObject runBySequentialComputing() { try { //计时器工具初始化 ProgramTimer pt = new ProgramTimer(); pt.start(); //IP分布范围工具初始化 IpAreaRange ipar = new IpAreaRange(); //MAC地址统计工具初始化 MacAddressStatistics mast = new MacAddressStatistics(); //日志格式正则(NginX日志,txt) //pattern = "([^ ]*) ([^ ]*) ([^ ]*) (\\[.*\\]) (\\\".*?\\\") (-|[0-9]*) (-|[0-9]*) (\\\".*?\\\") (\\\".*?\\\") \\\"([^\\\"]*)\\\""; //日志格式正则(比赛日志,csv) String pattern = "(,)?((\\\"[^\\\"]*(\\\"{2})*[^\\\"]*\\\")*[^,]*)"; Pattern r = Pattern.compile(pattern); //读入日志文件 File file = new File(FilePath.logFile); //缓冲流 BufferedInputStream fis = new BufferedInputStream(new FileInputStream(file)); //缓冲区大小为5M BufferedReader reader = new BufferedReader(new InputStreamReader(fis, "utf-8"), 5 * 1024 * 1024); String line = ""; //处理的记录数 int dataCount = 0; int sumDataCount = 0; //提取的条件 String needDate = "2018-07-17"; //串行实现计算任务 while ((line = reader.readLine()) != null) { //抽取日志文件的一行进行处理 sumDataCount++; //匹配CSV中最小单位的数据 Matcher matcher = r.matcher(line); ArrayList listOfLineData = new ArrayList(); while(matcher.find()) { String cell = matcher.group(2);//group(2) is ((\"[^\"]*(\"{2})*[^\"]*\")*[^,]*) Pattern pattern2 = Pattern.compile("\"((.)*)\""); Matcher matcher2 = pattern2.matcher(cell); if(matcher2.find()) { cell = matcher2.group(1); } //合并同一行的数据 listOfLineData.add(cell); } //按照匹配条件提取当前行需要的数据 if (listOfLineData.get(1).equals(needDate)){ String region; //该记录存在多个IP地址 if (listOfLineData.get(5).contains(",")){ String[] tmp = listOfLineData.get(5).split(", "); region = ipar.search(tmp[tmp.length-1]); } else { region = ipar.search(listOfLineData.get(5)); } //该记录不存在MAC地址 if (listOfLineData.get(4).isEmpty()){ mast.insert("00-00-00-00-00-00"); } else { mast.insert(listOfLineData.get(4)); } dataCount++; } } //输出匹配记录数 System.out.println("Matched/Summary Data Counter: " + dataCount + "/" + sumDataCount); //程序运行时间 System.out.println("串行计算" + pt.runningTime()); //将结果转换为JSON传出 JSONArray arrayOfMac = JSONArray.fromObject(mast.getAnswer()); JSONArray arrayOfIp = JSONArray.fromObject(ipar.getRegionCount()); JSONObject jsonOfAnswer = new JSONObject(); jsonOfAnswer.put("mac",arrayOfMac); jsonOfAnswer.put("prov",arrayOfIp); jsonOfAnswer.put("Matched",dataCount); jsonOfAnswer.put("Summary",sumDataCount); jsonOfAnswer.put("RunningTime:",pt.runningTime()); return jsonOfAnswer; } catch (IOException e) { System.out.println(e); } return null; } /** * 程序主入口-计算核心的并行实现-线程安全 * * return jsonOfAnswer JSONObject */ public static JSONObject runByParallelComputing() { try { //计时器工具初始化 ProgramTimer pt = new ProgramTimer(); pt.start(); //IP分布范围工具初始化 IpAreaRange ipar = new IpAreaRange(); //MAC地址统计工具初始化 MacAddressStatistics mast = new MacAddressStatistics(); //日志格式正则(NginX日志,txt) //pattern = "([^ ]*) ([^ ]*) ([^ ]*) (\\[.*\\]) (\\\".*?\\\") (-|[0-9]*) (-|[0-9]*) (\\\".*?\\\") (\\\".*?\\\") \\\"([^\\\"]*)\\\""; //日志格式正则(比赛日志,csv) String pattern = "(,)?((\\\"[^\\\"]*(\\\"{2})*[^\\\"]*\\\")*[^,]*)"; //正则匹配初始化 Pattern r = Pattern.compile(pattern); //读入日志文件 File file = new File(FilePath.logFile); //缓冲流 BufferedInputStream fis = new BufferedInputStream(new FileInputStream(file)); //缓冲区大小为5M BufferedReader reader = new BufferedReader(new InputStreamReader(fis, "utf-8"), 5 * 1024 * 1024); //处理的记录数(总记录数) int[] testSumDataCount = {0}; //处理的记录数(按照条件匹配到的记录数) int[] testDataCount = {0}; //提取的条件 String needDate = "2018-07-17"; //线程锁 Lock lock = new ReentrantLock(); //根据物理机CPU核心数自动判断可以开启的线程数量,并行处理计算任务 reader.lines().parallel().forEach((String line) -> { //每个线程抽取日志文件的一行进行处理 //原子操作使用线程锁确保线程安全 lock.lock(); try{ testSumDataCount[0] ++ ; } finally { lock.unlock(); } //匹配CSV中最小单位的数据 Matcher matcher = r.matcher(line); List listOfLineData = Collections.synchronizedList(new ArrayList()); while(matcher.find()) { String cell = matcher.group(2);//group(2) is ((\"[^\"]*(\"{2})*[^\"]*\")*[^,]*) Pattern pattern2 = Pattern.compile("\"((.)*)\""); Matcher matcher2 = pattern2.matcher(cell); if(matcher2.find()) { cell = matcher2.group(1); } //合并同一行的数据 listOfLineData.add(cell); } //按照匹配条件提取当前行需要的数据 if (listOfLineData.get(1).equals(needDate)){ String region; String infoOfIp; String infoOfMac; //该记录存在多个IP地址 if (listOfLineData.get(5).contains(",")){ String[] tmp = listOfLineData.get(5).split(", "); infoOfIp = tmp[tmp.length-1]; } else { infoOfIp = listOfLineData.get(5); } //该记录不存在MAC地址 if (listOfLineData.get(4).isEmpty()){ infoOfMac = "00-00-00-00-00-00"; } else { infoOfMac = listOfLineData.get(4); } //对取得的数据进行记录(线程安全) lock.lock(); try{ ipar.search(infoOfIp); mast.insert(infoOfMac); testDataCount[0]++; } finally { lock.unlock(); } } }); //输出匹配记录数 System.out.println("Matched/Summary Data Counter: " + testDataCount[0] + "/" + testSumDataCount[0]); //程序运行时间 System.out.println("并行计算" + pt.runningTime()); //将结果转换为JSON传出 JSONArray arrayOfMac = JSONArray.fromObject(mast.getAnswer()); JSONArray arrayOfIp = JSONArray.fromObject(ipar.getRegionCount()); JSONObject jsonOfAnswer = new JSONObject(); jsonOfAnswer.put("mac",arrayOfMac); jsonOfAnswer.put("prov",arrayOfIp); jsonOfAnswer.put("Matched",testDataCount[0]); jsonOfAnswer.put("Summary",testSumDataCount[0]); jsonOfAnswer.put("RunningTime:",pt.runningTime()); return jsonOfAnswer; } catch (IOException e) { System.out.println(e); } return null; } /** * 程序主入口-计算核心的并行实现-非线程安全 * * return jsonOfAnswer JSONObject */ public static JSONObject runByParallelComputingNotSafe() { try { //计时器工具初始化 ProgramTimer pt = new ProgramTimer(); pt.start(); //IP分布范围工具初始化 IpAreaRange ipar = new IpAreaRange(); //MAC地址统计工具初始化 MacAddressStatistics mast = new MacAddressStatistics(); //日志格式正则(NginX日志,txt) //pattern = "([^ ]*) ([^ ]*) ([^ ]*) (\\[.*\\]) (\\\".*?\\\") (-|[0-9]*) (-|[0-9]*) (\\\".*?\\\") (\\\".*?\\\") \\\"([^\\\"]*)\\\""; //日志格式正则(比赛日志,csv) String pattern = "(,)?((\\\"[^\\\"]*(\\\"{2})*[^\\\"]*\\\")*[^,]*)"; //正则匹配初始化 Pattern r = Pattern.compile(pattern); //读入日志文件 File file = new File(FilePath.logFile); //缓冲流 BufferedInputStream fis = new BufferedInputStream(new FileInputStream(file)); //缓冲区大小为5M BufferedReader reader = new BufferedReader(new InputStreamReader(fis, "utf-8"), 5 * 1024 * 1024); //处理的记录数(总记录数) int[] testSumDataCount = {0}; //处理的记录数(按照条件匹配到的记录数) int[] testDataCount = {0}; //提取的条件 String needDate = "2018-07-17"; //线程锁 Lock lock = new ReentrantLock(); //根据物理机CPU核心数自动判断可以开启的线程数量,并行处理计算任务 reader.lines().parallel().forEach((String line) -> { //每个线程抽取日志文件的一行进行处理 //原子操作使用线程锁确保线程安全 lock.lock(); try{ testSumDataCount[0] ++ ; } finally { lock.unlock(); } //匹配CSV中最小单位的数据 Matcher matcher = r.matcher(line); List listOfLineData = Collections.synchronizedList(new ArrayList()); while(matcher.find()) { String cell = matcher.group(2);//group(2) is ((\"[^\"]*(\"{2})*[^\"]*\")*[^,]*) Pattern pattern2 = Pattern.compile("\"((.)*)\""); Matcher matcher2 = pattern2.matcher(cell); if(matcher2.find()) { cell = matcher2.group(1); } //合并同一行的数据 listOfLineData.add(cell); } //按照匹配条件提取当前行需要的数据 if (listOfLineData.get(1).equals(needDate)){ String region; String infoOfIp; String infoOfMac; //该记录存在多个IP地址 if (listOfLineData.get(5).contains(",")){ String[] tmp = listOfLineData.get(5).split(", "); infoOfIp = tmp[tmp.length-1]; } else { infoOfIp = listOfLineData.get(5); } //该记录不存在MAC地址 if (listOfLineData.get(4).isEmpty()){ infoOfMac = "00-00-00-00-00-00"; } else { infoOfMac = listOfLineData.get(4); } ipar.search(infoOfIp); mast.insert(infoOfMac); lock.lock(); try{ testDataCount[0]++; } finally { lock.unlock(); } } }); //输出匹配记录数 System.out.println("Matched/Summary Data Counter: " + testDataCount[0] + "/" + testSumDataCount[0]); //程序运行时间 System.out.println("非安全并行计算" + pt.runningTime()); //将结果转换为JSON传出 JSONArray arrayOfMac = JSONArray.fromObject(mast.getAnswer()); JSONArray arrayOfIp = JSONArray.fromObject(ipar.getRegionCount()); JSONObject jsonOfAnswer = new JSONObject(); jsonOfAnswer.put("mac",arrayOfMac); jsonOfAnswer.put("prov",arrayOfIp); jsonOfAnswer.put("Matched",testDataCount[0]); jsonOfAnswer.put("Summary",testSumDataCount[0]); jsonOfAnswer.put("RunningTime:",pt.runningTime()); return jsonOfAnswer; } catch (IOException e) { System.out.println(e); } return null; } /** * 获取系统性能信息的入口函数(注意!仅支持Windows系统和Linux系统,不支持MacOS) * * @return jsonOfAnswer JSONObject */ public static JSONObject getSystemInfo() throws Exception { //当前系统的CPU使用率 double cpuUsage = SystemInfo.getCpuUsage(); //当前系统的内存使用率 double memUsage = SystemInfo.getMemUsage(); //当前系统的硬盘使用率 double diskUsage = SystemInfo.getDiskUsage(); JSONObject jsonOfAnswer = new JSONObject(); jsonOfAnswer.put("CPU",Integer.toString( Integer.parseInt(new java.text.DecimalFormat("0").format(cpuUsage)))); return jsonOfAnswer; } public static void main(String[] args){ //System.out.println(MainDataAnalysis.runBySequentialComputing()); //System.out.println(MainDataAnalysis.runByParallelComputing()); System.out.println(MainDataAnalysis.runByParallelComputingNotSafe()); } }