线程统一启动

潘钊
1 parent 228a199d
Showing 5 changed files with 33 additions and 55 deletions
src/main/java/com/bsth/XDApplication.java
src/main/java/com/bsth/data/BasicData.java
src/main/java/com/bsth/data/forecast/ForecastRealServer.java
src/main/java/com/bsth/data/msg_queue/DirectivePushQueue.java
src/main/java/com/bsth/data/msg_queue/WebSocketPushQueue.java
@@ -4,8 +4,11 @@ import com.bsth.data.BasicData;
 import com.bsth.data.ThreadMonotor;
 import com.bsth.data.car_out_info.UpdateDBThread;
 import com.bsth.data.directive.DirectivesPstThread;
+import com.bsth.data.forecast.SampleTimeDataLoader;
 import com.bsth.data.gpsdata.thread.GpsDataLoaderThread;
 import com.bsth.data.gpsdata.thread.OfflineMonitorThread;
+import com.bsth.data.msg_queue.DirectivePushQueue;
+import com.bsth.data.msg_queue.WebSocketPushQueue;
 import com.bsth.data.schedule.edit_logs.SeiPstThread;
 import com.bsth.data.schedule.late_adjust.ScheduleLateThread;
 import com.bsth.data.schedule.thread.CalcOilThread;
@@ -56,6 +59,8 @@ public class XDApplication implements CommandLineRunner {
     ThreadMonotor threadMonotor;
     @Autowired
     SeiPstThread seiPstThread;
+    @Autowired
+    SampleTimeDataLoader sampleTimeDataLoader;
     private static long timeDiff;
     private static long timeDiffTraffic;
@@ -102,7 +107,7 @@ public class XDApplication implements CommandLineRunner {
         //dayOfSchedule.dataRecovery();
         //sexec.scheduleWithFixedDelay(gpsDataLoader, 30, 2, TimeUnit.SECONDS);
         //实际排班更新线程
-        //sexec.scheduleWithFixedDelay(scheduleRefreshThread, 15, 240, TimeUnit.SECONDS);
+        sexec.scheduleWithFixedDelay(scheduleRefreshThread, 15, 240, TimeUnit.SECONDS);
         //实际排班延迟入库线程
         //sexec.scheduleWithFixedDelay(schedulePstThread, 60, 15, TimeUnit.SECONDS);
         //班次修正日志延迟入库
@@ -116,30 +121,32 @@ public class XDApplication implements CommandLineRunner {
     public void prodInit(){
         log.info("prodInit...");
         ScheduledExecutorService sexec = Application.mainServices;
-        //发车信息
-        sexec.scheduleWithFixedDelay(fcxxUpdateThread, 60, 40, TimeUnit.SECONDS);
-        //抓取GPS数据
-        sexec.scheduleWithFixedDelay(gpsDataLoader, 30, 2, TimeUnit.SECONDS);
-        //GPS设备掉离线
-        sexec.scheduleWithFixedDelay(offlineMonitorThread, 120, 60, TimeUnit.SECONDS);
-        //班次更新线程
-        sexec.scheduleWithFixedDelay(scheduleRefreshThread, 15, 240, TimeUnit.SECONDS);
-        //班次延迟入库线程
-        sexec.scheduleWithFixedDelay(schedulePstThread, 60, 10, TimeUnit.SECONDS);
-        //班次修正日志入库
-        sexec.scheduleWithFixedDelay(seiPstThread, 60, 60, TimeUnit.SECONDS);
-        //检查班次误点
-        sexec.scheduleWithFixedDelay(scheduleLateThread, 60, 30, TimeUnit.SECONDS);
-        //调度指令延迟入库
-        sexec.scheduleWithFixedDelay(directivesPstThread, 180, 100, TimeUnit.SECONDS);
+
+        /** 线调业务 */
+        sexec.scheduleWithFixedDelay(gpsDataLoader, 30, 2, TimeUnit.SECONDS);//抓取GPS数据
+        sexec.scheduleWithFixedDelay(offlineMonitorThread, 120, 60, TimeUnit.SECONDS);//GPS设备掉离线
+        sexec.scheduleWithFixedDelay(scheduleRefreshThread, 15, 240, TimeUnit.SECONDS);//班次更新线程
+        sexec.scheduleWithFixedDelay(schedulePstThread, 60, 10, TimeUnit.SECONDS);//班次延迟入库线程
+        sexec.scheduleWithFixedDelay(seiPstThread, 60, 60, TimeUnit.SECONDS);//班次修正日志入库
+        sexec.scheduleWithFixedDelay(scheduleLateThread, 60, 30, TimeUnit.SECONDS);//检查班次误点
+        sexec.scheduleWithFixedDelay(directivesPstThread, 180, 100, TimeUnit.SECONDS);//调度指令延迟入库
+        sexec.scheduleWithFixedDelay(threadMonotor, 240, 60, TimeUnit.SECONDS);//线程监听（防止重要的线程阻塞、异常结束。以及部分主备切换工作）
+        sexec.scheduleWithFixedDelay(sampleTimeDataLoader, 12, 120 * 60, TimeUnit.SECONDS);//到离站预测需要的站点间耗时数据
+        sexec.scheduleWithFixedDelay(basicDataLoader, 2, 2, TimeUnit.HOURS);//基础数据更新
+        DirectivePushQueue.start();//消息队列 -指令，系统下发的
+        WebSocketPushQueue.start();//消息队列 -webSocket ，推送至线调web页面的
+
+        /** 线调为其他程序提供的数据 --写入数据库 */
+        sexec.scheduleWithFixedDelay(fcxxUpdateThread, 60, 40, TimeUnit.SECONDS);//发车信息（发车屏、信息发布）
+        //线路首末班数据（网关用，班次更新时写入）
+        //com.bsth.data.schedule.f_a_l.FirstAndLastHandler
+        //无法自动完成的班次信息（网关用，补信号）
+
         //运管处静态数据提交
         log.info(timeDiff / 1000 / 60 + "分钟之后提交到运管处");
         sexec.scheduleAtFixedRate(submitToTrafficManage, timeDiffTraffic / 1000, 60 * 60 * 24, TimeUnit.SECONDS);
         //计算油、公里加注
         sexec.scheduleAtFixedRate(calcOilThread, timeDiff / 1000, 60 * 60 * 24, TimeUnit.SECONDS);
-
-        //线程监听（防止重要的线程阻塞、异常结束。以及部分主备切换工作）
-        sexec.scheduleWithFixedDelay(threadMonotor, 240, 60, TimeUnit.SECONDS);
     }
 }
 package com.bsth.data;
-import com.bsth.Application;
 import com.bsth.data.gpsdata.arrival.GeoCacheData;
 import com.bsth.data.pinyin.PersionPinYinBuffer;
 import com.bsth.entity.*;
@@ -13,12 +12,10 @@ import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.CommandLineRunner;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 import java.util.*;
-import java.util.concurrent.TimeUnit;
 /**
  * @author PanZhao
@@ -27,7 +24,7 @@ import java.util.concurrent.TimeUnit;
  * @date 2016年8月10日 下午3:27:45
  */
 @Component
-public class BasicData implements CommandLineRunner {
+public class BasicData {
     //公司代码和公司名对照(K: 公司编码，V:公司名)
     public static Map<String, String> businessCodeNameMap;
@@ -80,14 +77,6 @@ public class BasicData implements CommandLineRunner {
     static Logger logger = LoggerFactory.getLogger(BasicData.class);
-    @Autowired
-    BasicDataLoader dataLoader;
-
-    @Override
-    public void run(String... arg0) throws Exception {
-        Application.mainServices.scheduleWithFixedDelay(dataLoader, 2, 2, TimeUnit.HOURS);
-    }
-
     public static String getStationNameByCode(String code, String prefix){
         String name = stationCode2NameMap.get(code);
         return name != null? name: stationCode2NameMap.get(prefix + code);
 package com.bsth.data.forecast;
-import com.bsth.Application;
 import com.bsth.data.forecast.entity.ForecastResult;
 import com.bsth.data.forecast.entity.ForecastResult.ForecastResultItem;
 import com.bsth.data.forecast.entity.SimpleRoute;
@@ -13,7 +12,6 @@ import com.google.common.collect.ArrayListMultimap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
 import java.text.DecimalFormat;
@@ -21,7 +19,6 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 /**
  * 
@@ -32,10 +29,7 @@ import java.util.concurrent.TimeUnit;
  *
  */
 @Component
-public class ForecastRealServer implements CommandLineRunner {
-	
-	@Autowired
-	SampleTimeDataLoader dataLoader;
+public class ForecastRealServer {
 	@Autowired
 	DayOfSchedule dayOfSchedule;
@@ -57,12 +51,6 @@ public class ForecastRealServer implements CommandLineRunner {
 		forecastMap = new HashMap<>();
 	}
-	@Override
-	public void run(String... arg0) throws Exception {
-		//2小时更新一次站点间耗时数据
-		Application.mainServices.scheduleWithFixedDelay(dataLoader, 12, 120 * 60, TimeUnit.SECONDS);
-	}
-	
 	/**
 	 * 
 	 * @Title: forecast 
@@ -5,7 +5,6 @@ import com.bsth.service.directive.DirectiveService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
-import org.springframework.boot.CommandLineRunner;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Component;
@@ -17,7 +16,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
  * Created by panzhao on 2017/5/11.
  */
 @Component
-public class DirectivePushQueue implements CommandLineRunner, ApplicationContextAware {
+public class DirectivePushQueue implements ApplicationContextAware {
     static ConcurrentLinkedQueue<QueueData_Directive> linkedList;
     static DataPushThread thread;
@@ -80,10 +79,10 @@ public class DirectivePushQueue implements CommandLineRunner, ApplicationContext
         return linkedList.size();
     }
-    @Override
+/*    @Override
     public void run(String... strings) throws Exception {
         start();
-    }
+    }*/
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -3,7 +3,6 @@ package com.bsth.data.msg_queue;
 import com.bsth.common.Constants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
@@ -15,7 +14,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
  * Created by panzhao on 2017/5/11.
  */
 @Component
-public class WebSocketPushQueue implements CommandLineRunner {
+public class WebSocketPushQueue {
     static ConcurrentLinkedQueue<QueueData> linkedList;
     static DataPushThread thread;
@@ -54,10 +53,6 @@ public class WebSocketPushQueue implements CommandLineRunner {
         return linkedList.size();
     }
-    @Override
-    public void run(String... strings) throws Exception {
-        start();
-    }
     public static class DataPushThread extends Thread {