update...

潘钊
1 parent 72fd6f2d
Showing 1 changed file with 72 additions and 47 deletions
src/main/java/com/bsth/data/gpsdata_v2/DataHandleProcess.java
@@ -4,18 +4,14 @@ import com.bsth.data.gpsdata_v2.cache.GpsCacheData;
 import com.bsth.data.gpsdata_v2.entity.GpsEntity;
 import com.bsth.data.gpsdata_v2.handlers.*;
 import com.google.common.collect.ArrayListMultimap;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
+import java.util.*;
+import java.util.concurrent.*;
 /**
  * 实时信号数据处理
@@ -24,12 +20,6 @@ import java.util.concurrent.ThreadFactory;
 @Component
 public class DataHandleProcess {
-    static Logger logger = LoggerFactory.getLogger(DataHandleProcess.class);
-    final static int POOL_SIZE = 25;
-
-    static ExecutorService threadPool = Executors.newFixedThreadPool(POOL_SIZE + 1, new HandlerThreadFactory());
-    public static CountDownLatch count;
-
     @Autowired
     GpsStateProcess gpsStateProcess;
     @Autowired
@@ -42,10 +32,17 @@ public class DataHandleProcess {
     OutStationProcess outStationProcess;
     @Autowired
     ReverseRouteProcess reverseRouteProcess;
-
     @Autowired
     GpsRealData gpsRealData;
+
+    static Logger logger = LoggerFactory.getLogger(DataHandleProcess.class);
+
+    final static int POOL_SIZE = 25;
+
+    static ExecutorService threadPool = Executors.newFixedThreadPool(POOL_SIZE + 1);
+    public static CountDownLatch count;
+
     static long lastTime;
     public static boolean isBlock() {
@@ -54,7 +51,7 @@ public class DataHandleProcess {
     public void handle(List<GpsEntity> list) {
         try {
-            if(list.size() == 0)
+            if (list.size() == 0)
                 return;
             lastTime = System.currentTimeMillis();
             //按设备号分组数据（一个设备的多条数据，必须在一个线程里跑）
@@ -76,21 +73,50 @@ public class DataHandleProcess {
             logger.info("analyse gps size: " + list.size() + ", ks: " + ks.size());
             count = new CountDownLatch(ks.size());
+            List<Future> fRs = new ArrayList<>(ks.size());
             for (Integer index : ks) {
-                threadPool.execute(new SignalHandleThread(dataListMap.get(index), count));
+                fRs.add(threadPool.submit(new SignalHandleThread(dataListMap.get(index), count)));
             }
+            //按线路分组gps
+            /*ArrayListMultimap multimap = ArrayListMultimap.create();
+            for (GpsEntity gps : list) {
+                multimap.put(gps.getLineId(), gps);
+            }
+
+            Set<String> ks = multimap.keySet();
+
+            logger.info("analyse gps size: " + list.size() + ", ks: " + ks.size());
+            count = new CountDownLatch(ks.size());
+
+            for (String lineCode : ks) {
+                threadPool.execute(new SignalHandleThread(multimap.get(lineCode), count));
+            }*/
+
             //等待子线程结束
             count.await();
+            for (Future f : fRs) {
+                try {
+                    f.get();
+                } catch (InterruptedException e) {
+                } catch (ExecutionException e) {
+                    logger.error(e.getCause().getMessage());
+                }
+            }
+
             //加入实时gps对照
             for (GpsEntity gps : list)
                 gpsRealData.put(gps);
+
+            logger.info("time , " + (System.currentTimeMillis() - lastTime));
         } catch (Exception e) {
             logger.error("", e);
         }
     }
+    static GpsComp comp = new GpsComp();
+
     public class SignalHandleThread implements Runnable {
         List<GpsEntity> list;
@@ -103,45 +129,44 @@ public class DataHandleProcess {
         @Override
         public void run() {
-            try {
-                for (GpsEntity gps : list) {
-                    try{
-                        if(Math.abs(gps.getTimestamp() - gps.getServerTimestamp()) > 1000 * 60 * 20)
-                            continue;
-
-                        gpsStateProcess.process(gps);//状态处理
-                        stationInsideProcess.process(gps);//场站内外判定
-                        reverseRouteProcess.process(gps);//反向路由处理
-                        abnormalStateProcess.process(gps);//超速越界
-
-                        inStationProcess.process(gps);//进站
-                        outStationProcess.process(gps);//出站
-
-                        GpsCacheData.putGps(gps);//历史gps缓存
-                    }catch (Exception e){
-                        logger.error("", e);
-                    }
+            //try {
+            Collections.sort(list, comp);
+            for (GpsEntity gps : list) {
+                try {
+                    if(StringUtils.isEmpty(gps.getNbbm()))
+                        continue;
+                    if (Math.abs(gps.getTimestamp() - gps.getServerTimestamp()) > 1000 * 60 * 20)
+                        continue;
+
+                    gpsStateProcess.process(gps);//状态处理
+                    stationInsideProcess.process(gps);//场站内外判定
+                    reverseRouteProcess.process(gps);//反向路由处理
+                    abnormalStateProcess.process(gps);//超速越界
+
+                    inStationProcess.process(gps);//进站
+                    outStationProcess.process(gps);//出站
+
+                    GpsCacheData.putGps(gps);//历史gps缓存
+                } catch (Exception e) {
+                    logger.error("", e);
                 }
-            } finally {
+            }
+
+            logger.info(Thread.currentThread().getName() + " -countDown : " + count.getCount());
+            count.countDown();
+            /*} finally {
                 if (count != null)
                     count.countDown();
-            }
+                logger.info(Thread.currentThread().getName() + " -countDown : " + count.getCount());
+            }*/
         }
     }
-    static class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
-        @Override
-        public void uncaughtException(Thread t, Throwable e) {
-            logger.error("caught " , e);
-        }
-    }
+    public static class GpsComp implements Comparator<GpsEntity> {
-    static class HandlerThreadFactory implements ThreadFactory {
         @Override
-        public Thread newThread(Runnable r) {
-            Thread t = new Thread(r);
-            t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
-            return t;
+        public int compare(GpsEntity g1, GpsEntity g2) {
+            return g1.getTimestamp().compareTo(g2.getTimestamp());
         }
     }
 }
 \ No newline at end of file