update...

潘钊
1 parent d9cd62fc
Showing 9 changed files with 143 additions and 55 deletions
src/main/java/com/bsth/controller/realcontrol/anomalyCheckController.java
src/main/java/com/bsth/data/directive/GatewayHttpUtils.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/thread/GpsDataLoaderThread.java
src/main/java/com/bsth/data/gpsdata/thread/ThreadPollMonitor.java
src/main/java/com/bsth/data/schedule/DayOfSchedule.java
src/main/java/com/bsth/data/schedule/ScheduleComparator.java
src/main/java/com/bsth/websocket/handler/RealControlSocketHandler.java
 package com.bsth.controller.realcontrol;
+import com.bsth.data.gpsdata.arrival.GpsRealAnalyse;
 import com.bsth.data.schedule.DayOfSchedule;
 import com.bsth.entity.realcontrol.ScheduleRealInfo;
 import org.slf4j.Logger;
@@ -49,4 +50,9 @@ public class anomalyCheckController {
             dayOfSchedule.replaceByNbbm(nbbm, map.values());
         }
     }
+
+    @RequestMapping(value = "/shutdownThreadPool")
+    public void shutdownThreadPool(){
+        GpsRealAnalyse.shutdown();
+    }
 }
 package com.bsth.data.directive;
-import java.io.IOException;
-
+import com.alibaba.fastjson.JSONObject;
+import com.bsth.util.ConfigUtil;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
@@ -12,8 +12,7 @@ import org.apache.http.util.EntityUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import com.alibaba.fastjson.JSONObject;
-import com.bsth.util.ConfigUtil;
+import java.io.IOException;
 /**
  * 
@@ -42,7 +41,7 @@ public class GatewayHttpUtils {
 			//超时时间
 			RequestConfig requestConfig = RequestConfig.custom()  
-			        .setConnectTimeout(3000).setConnectionRequestTimeout(1000)  
+			        .setConnectTimeout(3000).setConnectionRequestTimeout(1000)
 			        .setSocketTimeout(3000).build();
 			HttpPost post = new HttpPost(url);
@@ -4,6 +4,7 @@ import com.bsth.data.BasicData;
 import com.bsth.data.forecast.ForecastRealServer;
 import com.bsth.data.gpsdata.thread.GpsDataLoaderThread;
 import com.bsth.data.gpsdata.thread.OfflineMonitorThread;
+import com.bsth.data.gpsdata.thread.ThreadPollMonitor;
 import com.bsth.data.schedule.DayOfSchedule;
 import com.bsth.entity.realcontrol.ScheduleRealInfo;
 import com.google.common.collect.TreeMultimap;
@@ -46,6 +47,9 @@ public class GpsRealData implements CommandLineRunner {
     @Autowired
     ForecastRealServer forecastRealServer;
+    @Autowired
+    ThreadPollMonitor threadPollMonitor;
+
     /**
      * 构造函数
      */
@@ -61,6 +65,10 @@ public class GpsRealData implements CommandLineRunner {
         //Application.mainServices.scheduleWithFixedDelay(gpsDataLoader, 20, 2, TimeUnit.SECONDS);
         //定时扫描掉离线
         //Application.mainServices.scheduleWithFixedDelay(offlineMonitorThread, 120, 60, TimeUnit.SECONDS);
+
+        //扫描GPS线程池状态
+        //Application.mainServices.scheduleWithFixedDelay(threadPollMonitor, 60, 20, TimeUnit.SECONDS);
+
     }
@@ -26,7 +26,7 @@ import java.util.concurrent.Executors;
 @Component
 public class GpsRealAnalyse {
-    Logger logger = LoggerFactory.getLogger(this.getClass());
+    static Logger logger = LoggerFactory.getLogger(GpsRealAnalyse.class);
     @Autowired
     OfflineSignalHandle offlineSignalHandle;
@@ -46,40 +46,54 @@ public class GpsRealAnalyse {
     static ExecutorService threadPool = Executors.newFixedThreadPool(100);
+    public static long st;
+    public static CountDownLatch count;
+
     public void analyse(List<GpsEntity> list) {
-        //如果正在恢复数据
-        if (GpsDataRecovery.run)
-            return;
-
-        long t = System.currentTimeMillis();
-        logger.info("analyse gps size: " + list.size());
-        //按线路分组gps
-        ArrayListMultimap multimap = ArrayListMultimap.create();
-        for(GpsEntity gps : list){
-            multimap.put(gps.getLineId(), gps);
-        }
+        try {
+            st = System.currentTimeMillis();
+            //如果正在恢复数据
+            if (GpsDataRecovery.run)
+                return;
+
+            //按线路分组gps
+            ArrayListMultimap multimap = ArrayListMultimap.create();
+            for (GpsEntity gps : list) {
+                multimap.put(gps.getLineId(), gps);
+            }
-        Set<String> ks = multimap.keySet();
-        CountDownLatch count = new CountDownLatch(ks.size());
+            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));
-        }
+            for (String lineCode : ks) {
+                threadPool.execute(new SignalHandleThread(multimap.get(lineCode), count));
+            }
-        try {
             //等待子线程结束
             count.await();
             //加入实时gps对照
-            for(GpsEntity gps: list)
+            for (GpsEntity gps : list)
                 gpsRealData.put(gps);
-            logger.info("time , " + (System.currentTimeMillis() - t));
-        } catch (InterruptedException e) {
+            logger.info("time , " + (System.currentTimeMillis() - st));
+        } catch (Exception e) {
             logger.error("", e);
         }
     }
+    public static void shutdown() {
+        logger.warn("GpsRealAnalyse shutdown！！");
+        threadPool.shutdownNow();
+        long len = count.getCount();
+        for (int i = 0; i < len; i++) {
+            count.countDown();
+        }
+
+        threadPool = Executors.newFixedThreadPool(100);
+    }
+
     static GpsComp comp = new GpsComp();
     public class SignalHandleThread implements Runnable {
@@ -97,7 +111,7 @@ public class GpsRealAnalyse {
             try {
                 Collections.sort(list, comp);
-                for(GpsEntity gps : list){
+                for (GpsEntity gps : list) {
                     //是否有任务
                     boolean task;
                     CircleQueue<GpsEntity> prevs = GeoCacheData.getGps(gps.getNbbm());
@@ -110,7 +124,7 @@ public class GpsRealAnalyse {
                     //异常判定（越界/超速）
                     abnormalStateHandle.handle(gps, prevs);
-                    if(!task)
+                    if (!task)
                         return; //无任务的，到这里就结束
                     //反向处理
@@ -122,7 +136,7 @@ public class GpsRealAnalyse {
             } catch (Exception e) {
                 logger.error("", e);
             } finally {
-                if(count != null)
+                if (count != null)
                     count.countDown();
             }
         }
@@ -101,9 +101,9 @@ public class GpsDataLoaderThread extends Thread {
                     old = gpsRealData.get(gps.getDeviceId());
                     if (old != null &&
-                            old.getTimestamp() == gps.getTimestamp() &&
-                            old.getLat() == gps.getLat() &&
-                            old.getLon() == gps.getLon())
+                            old.getTimestamp().equals(gps.getTimestamp()) &&
+                            old.getLat().equals(gps.getLat()) &&
+                            old.getLon().equals(gps.getLon()))
                         continue;
                     nbbm = BasicData.deviceId2NbbmMap.get(gps.getDeviceId());
+package com.bsth.data.gpsdata.thread;
+
+import com.bsth.data.gpsdata.arrival.GpsRealAnalyse;
+import org.springframework.stereotype.Component;
+
+/**
+ * 线程池监听
+ * Created by panzhao on 2017/5/10.
+ */
+@Component
+public class ThreadPollMonitor extends Thread{
+
+    @Override
+    public void run() {
+        long t = System.currentTimeMillis();
+
+        if(t - GpsRealAnalyse.st > 3000 * 10){
+            GpsRealAnalyse.shutdown();
+        }
+    }
+}
@@ -833,6 +833,29 @@ public class DayOfSchedule implements CommandLineRunner {
         return sch;
     }
+    /**
+     * 搜索离当前时间最近的一个指定类型的班次
+     * @param nbbm
+     * @param bcType
+     * @return
+     */
+    public ScheduleRealInfo searchNearByBcType(String nbbm, String bcType){
+        List<ScheduleRealInfo> list = findByBcType(nbbm, bcType);
+        Collections.sort(list, schFCSJComparator);
+
+        long t = System.currentTimeMillis();
+        int distance=-1;
+
+        ScheduleRealInfo sch = null;
+        for(ScheduleRealInfo temp : list){
+
+            if(Math.abs(temp.getDfsjT() - t) < distance || distance == -1){
+                sch = temp;
+            }
+        }
+        return sch;
+    }
+
     public List<ScheduleRealInfo> findByBcType(String nbbm, String bcType) {
         List<ScheduleRealInfo> all = nbbmScheduleMap.get(nbbm), outList = new ArrayList<>();
 package com.bsth.data.schedule;
-import java.util.Comparator;
-
 import com.bsth.entity.realcontrol.ScheduleRealInfo;
+import java.util.Comparator;
+
 /**
  * 
  * @ClassName: ScheduleComparator 
@@ -25,7 +25,7 @@ public class ScheduleComparator {
 		@Override
 		public int compare(ScheduleRealInfo s1, ScheduleRealInfo s2) {
-			return (int) (s1.getDfsjT() - s2.getDfsjT());
+			return (int) (s1.getFcsjT() - s2.getFcsjT());
 		}
 	}
 }
@@ -4,7 +4,6 @@ import com.alibaba.fastjson.JSONObject;
 import com.bsth.common.Constants;
 import com.bsth.data.BasicData;
 import com.google.common.base.Splitter;
-import com.google.common.collect.ArrayListMultimap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.annotation.Scope;
@@ -12,6 +11,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.socket.*;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 /**
  * @author PanZhao
@@ -22,12 +22,14 @@ public class RealControlSocketHandler implements WebSocketHandler {
 	Logger logger = LoggerFactory.getLogger(this.getClass());
-	private static final ArrayList<WebSocketSession> users;
-	private static final ArrayListMultimap<String, WebSocketSession> listenMap;
+	private static ArrayList<WebSocketSession> users;
+	//private static final ArrayListMultimap<String, WebSocketSession> listenMap;
+	private static ConcurrentHashMap<String, List<WebSocketSession>> listenMap;
 	static {
 		users = new ArrayList<WebSocketSession>();
-		listenMap = ArrayListMultimap.create();
+		//listenMap = ArrayListMultimap.create();
+		listenMap = new ConcurrentHashMap();
 	}
 	@Override
@@ -35,19 +37,31 @@ public class RealControlSocketHandler implements WebSocketHandler {
 			throws Exception {
 		users.remove(session);
 		//清理监听
-		Set<String> keys = listenMap.keySet();
-		Map<String, WebSocketSession> remMap = new HashMap<>();
-		for(String k : keys){
-			if(listenMap.get(k).contains(session))
-				remMap.put(k, session);
-		}
+		//Set<String> keys = listenMap.keySet();
+		//Map<String, WebSocketSession> remMap = new HashMap<>();
+
+		int vsCount=0;
+		Collection<List<WebSocketSession>> vs = listenMap.values();
+		for(List<WebSocketSession> list : vs){
+			list.remove(session);
-		Set<String> remSet = remMap.keySet();
+			vsCount += list.size();
+		}
+		/*List<WebSocketSession> vs;
+		for(String k : keys){
+			//vs = listenMap.get(k);
+			//vs.remove(session);
+			listenMap.get(k).remove()
+			*//*if(listenMap.get(k).contains(session))
+				remMap.put(k, session);*//*
+		}*/
+
+		/*Set<String> remSet = remMap.keySet();
 		for(String k : remSet){
 			listenMap.remove(k, remMap.get(k));
 			logger.info("web socket close, remove listen K: "+ k);
-		}
-		logger.info("listen values size: " + listenMap.values().size() + " -CloseStatus:" + arg1);
+		}*/
+		logger.info("listen values size: " + vsCount + " -CloseStatus:" + arg1);
 	}
 	@Override
@@ -65,8 +79,12 @@ public class RealControlSocketHandler implements WebSocketHandler {
 				//注册线路监听
 				List<String> idx = Splitter.on(",").splitToList(jsonObj.getString("idx"));
 				for(String lineCode : idx){
-					if(BasicData.lineCode2NameMap.containsKey(lineCode))
-						listenMap.put(lineCode, session);
+					if(BasicData.lineCode2NameMap.containsKey(lineCode)){
+						if(!listenMap.containsKey(lineCode)){
+							listenMap.put(lineCode, new ArrayList<WebSocketSession>());
+						}
+						listenMap.get(lineCode).add(session);
+					}
 				}
 				break;
@@ -97,12 +115,11 @@ public class RealControlSocketHandler implements WebSocketHandler {
 	public void sendMessageToLine(String lineCode, String msg) {
 		TextMessage message = new TextMessage(msg.getBytes());
+		List<WebSocketSession> list = listenMap.get(lineCode);
+		if(list == null || list.size() == 0)
+			return;
-		Iterator<WebSocketSession> iterator = listenMap.get(lineCode).iterator();
-
-		WebSocketSession user;
-		while(iterator.hasNext()){
-			user = iterator.next();
+		for(WebSocketSession user : list){
 			try {
 				if (user.isOpen()) {
 					user.sendMessage(message);