update...

潘钊
1 parent aecfc182
Showing 6 changed files with 80 additions and 26 deletions
src/main/java/com/bsth/controller/realcontrol/anomalyCheckController.java
src/main/java/com/bsth/data/gpsdata/GpsRealData.java
src/main/java/com/bsth/data/gpsdata/arrival/GpsRealAnalyse.java
src/main/java/com/bsth/data/gpsdata/client/ClientApp.java
src/main/java/com/bsth/data/gpsdata/client/GpsBeforeBuffer.java
src/main/java/com/bsth/data/gpsdata/recovery/GpsDataRecovery.java
@@ -51,10 +51,17 @@ public class anomalyCheckController {
         }
     }
-    @RequestMapping(value = "/gpsClientReconn", method = RequestMethod.POST)
+    @Autowired
+    ClientApp clientApp;
+
+    @RequestMapping(value = "/gpsClientDestroy", method = RequestMethod.POST)
     public void gpsClientReconn(){
-        ClientApp.pdreconn();
-        ClientApp.pfreconn();
+        clientApp.destroy();
+    }
+
+    @RequestMapping(value = "/gpsClientInit", method = RequestMethod.POST)
+    public void gpsClientInit(){
+        clientApp.init();
     }
     @RequestMapping(value = "/pdClose", method = RequestMethod.POST)
@@ -4,6 +4,7 @@ import com.bsth.Application;
 import com.bsth.data.BasicData;
 import com.bsth.data.forecast.ForecastRealServer;
 import com.bsth.data.gpsdata.client.ClientApp;
+import com.bsth.data.gpsdata.client.GpsBeforeBuffer;
 import com.bsth.data.gpsdata.thread.GpsDataLoaderThread;
 import com.bsth.data.gpsdata.thread.OfflineMonitorThread;
 import com.bsth.data.schedule.DayOfSchedule;
@@ -58,6 +59,8 @@ public class GpsRealData implements CommandLineRunner {
     @Autowired
     ClientApp clientApp;
+    @Autowired
+    GpsBeforeBuffer gpsBeforeBuffer;
     @Override
     public void run(String... arg0) throws Exception {
         logger.info("gpsDataLoader,20,3");
@@ -68,6 +71,7 @@ public class GpsRealData implements CommandLineRunner {
         //gps 客户端
         //clientApp.init();
+        //gpsBeforeBuffer.init();
     }
@@ -11,6 +11,8 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -51,17 +53,17 @@ public class GpsRealAnalyse {
         long t = System.currentTimeMillis();
         logger.info("analyse gps size: " + list.size());
-        //按车辆分组gps
+        //按线路分组gps
         ArrayListMultimap multimap = ArrayListMultimap.create();
         for(GpsEntity gps : list){
-            multimap.put(gps.getNbbm(), gps);
+            multimap.put(gps.getLineId(), gps);
         }
         Set<String> ks = multimap.keySet();
         CountDownLatch count = new CountDownLatch(ks.size());
-        for(String nbbm : ks){
-            threadPool.execute(new SignalHandleThread(multimap.get(nbbm), count));
+        for(String lineCode : ks){
+            threadPool.execute(new SignalHandleThread(multimap.get(lineCode), count));
         }
         try {
@@ -78,6 +80,8 @@ public class GpsRealAnalyse {
         }
     }
+    static GpsComp comp = new GpsComp();
+
     public class SignalHandleThread implements Runnable {
         List<GpsEntity> list;
@@ -92,6 +96,7 @@ public class GpsRealAnalyse {
         public void run() {
             try {
+                Collections.sort(list, comp);
                 for(GpsEntity gps : list){
                     //是否有任务
                     boolean task;
@@ -122,4 +127,12 @@ public class GpsRealAnalyse {
             }
         }
     }
+
+    public static class GpsComp implements Comparator<GpsEntity> {
+
+        @Override
+        public int compare(GpsEntity g1, GpsEntity g2) {
+            return g1.getTimestamp().compareTo(g2.getTimestamp());
+        }
+    }
 }
@@ -42,16 +42,7 @@ public class ClientApp {
     static Logger logger = LoggerFactory.getLogger(ClientApp.class);
     private static ExecutorService exec;
-    private ScheduledExecutorService sexec = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
-
-        @Override
-        public Thread newThread(Runnable r) {
-            // TODO Auto-generated method stub
-            Thread t = new Thread(r);
-            t.setName("SessionCheckExecutor");
-            return t;
-        }
-    });
+    private ScheduledExecutorService sexec;
     public static boolean dconnect(String device) {
         boolean flag = false;
@@ -123,6 +114,20 @@ public class ClientApp {
         pfconnect(ConfigUtil.get("forward.device.name"));
     }
+    public void destroy(){
+        try {
+            logger.warn("socket client destroy!!!");
+            exec.shutdownNow();
+            sexec.shutdownNow();
+
+            pdDataConnector.dispose(true);
+            pfDataConnector.dispose(true);
+        } catch (Exception e) {
+            logger.error("", e);
+        }
+    }
+
+
     public static void pdClose(){
         pdSession.closeNow();
     }
@@ -149,7 +154,17 @@ public class ClientApp {
     }
     public void init() {
+        logger.warn("socket client init...");
         exec = Executors.newFixedThreadPool(50);
+        sexec = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+            @Override
+            public Thread newThread(Runnable r) {
+                // TODO Auto-generated method stub
+                Thread t = new Thread(r);
+                t.setName("SessionCheckExecutor");
+                return t;
+            }
+        });
         sexec.scheduleAtFixedRate(new SessionChecker(), 1, 1, TimeUnit.MINUTES);
         /*******************************浦东********************************/
         pdDataConnector = new NioSocketConnector();
@@ -192,9 +207,6 @@ public class ClientApp {
         pfDataConnector.setHandler(pfClient);
         pfconnect(ConfigUtil.get("forward.device.name"));
-
-
-        gpsBeforeBuffer.init();
     }
@@ -6,6 +6,8 @@ import com.bsth.data.gpsdata.GpsEntity;
 import com.bsth.data.gpsdata.arrival.GpsRealAnalyse;
 import com.bsth.data.gpsdata.client.pd.protocol.BasicInfo;
 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;
@@ -66,6 +68,7 @@ public class GpsBeforeBuffer {
         Application.mainServices.scheduleWithFixedDelay(gpsHandleThread, 20 * 1000, 1200, TimeUnit.MILLISECONDS);
     }
+    static int idleCount = 0;
     @Component
     public static class GpsHandleThread extends Thread{
@@ -74,6 +77,11 @@ public class GpsBeforeBuffer {
         @Autowired
         GpsRealAnalyse gpsRealAnalyse;
+        @Autowired
+        ClientApp clientApp;
+
+        Logger log = LoggerFactory.getLogger(this.getClass());
+
         @Override
         public void run() {
             list = new ArrayList<>(200);
@@ -86,6 +94,16 @@ public class GpsBeforeBuffer {
                 list.add(gps);
             }
+            if(list.size() == 0){
+                idleCount ++;
+                //连续40次没有数据，重建socket连接
+                if(idleCount == 40){
+                    log.info("idleCount == 40");
+                    idleCount = 0;
+                    clientApp.destroy();
+                    clientApp.init();
+                }
+            }
             gpsRealAnalyse.analyse(list);
         }
     }
@@ -45,11 +45,11 @@ public class GpsDataRecovery implements ApplicationContextAware {
     public void recovery() {
         List<GpsEntity> list = loadData();
-        //按车辆分组数据
+        //按线路分组数据
         ArrayListMultimap<String, GpsEntity> listMap = ArrayListMultimap.create();
         for (GpsEntity gps : list) {
-            if (gps.getNbbm() != null)
-                listMap.put(gps.getNbbm(), gps);
+            if (gps.getLineId() != null)
+                listMap.put(gps.getLineId(), gps);
         }
@@ -57,9 +57,9 @@ public class GpsDataRecovery implements ApplicationContextAware {
         CountDownLatch count = new CountDownLatch(keys.size());
         GpsComp comp = new GpsComp();
-        for (String nbbm : keys) {
-            Collections.sort(listMap.get(nbbm), comp);
-            threadPool.execute(new RecoveryThread(listMap.get(nbbm), count));
+        for (String lineId : keys) {
+            Collections.sort(listMap.get(lineId), comp);
+            threadPool.execute(new RecoveryThread(listMap.get(lineId), count));
             /*if(nbbm.equals("W7C-001"))
                 new RecoveryThread(listMap.get(nbbm), count).run();*/
         }