update...

潘钊
1 parent 7a7c9b2a
Showing 11 changed files with 79 additions and 36 deletions
src/main/java/com/bsth/data/gpsdata_v2/DataHandleProcess.java
src/main/java/com/bsth/data/gpsdata_v2/handlers/InStationProcess.java
src/main/java/com/bsth/data/gpsdata_v2/rfid/handle/RfidSignalHandle.java
src/main/java/com/bsth/data/gpsdata_v2/status_manager/gps_service_state/ServiceStateHandle.java
src/main/java/com/bsth/data/msg_queue/DirectivePushQueue.java
src/main/java/com/bsth/data/msg_queue/QueueData_Directive.java
src/main/java/com/bsth/data/pilot80/PilotReport.java
src/main/java/com/bsth/data/schedule/DayOfSchedule.java
src/main/java/com/bsth/service/directive/DirectiveService.java
src/main/java/com/bsth/service/directive/DirectiveServiceImpl.java
src/main/resources/static/real_control_v2/fragments/north/nav/directive_history.html
@@ -67,6 +67,8 @@ public class DataHandleProcess {
             //数据均分给线程
             ArrayListMultimap dataListMap = ArrayListMultimap.create();
             int size = deviceList.size(), threadIndex = 0, threadSize = size / POOL_SIZE;
+            if(threadSize==0)
+                threadSize = size;
             for (int i = 0; i < size; i++) {
                 if (i % threadSize == 0)
                     threadIndex++;
@@ -78,7 +80,6 @@ public class DataHandleProcess {
             logger.info(JSON.toJSONString(ks));
             for (Integer index : ks) {
-                //logger.info("execute index: " + index);
                 threadPool.execute(new SignalHandleThread(dataListMap.get(index), count));
             }
@@ -112,20 +113,17 @@ public class DataHandleProcess {
         public void run() {
             try {
                 Collections.sort(list, comp);
-                //logger.info("sort end --" + Thread.currentThread().getName() + " -list size: " + list.size());
-
                 GpsEntity gps;
                 for(int i = 0,len = list.size(); i< len ;i ++){
                     gps = list.get(i);
-                    //logger.info("run |--" + Thread.currentThread().getName() + " -i: " + i);
                     try {
                         if (StringUtils.isEmpty(gps.getNbbm()))
                             continue;
                         if (Math.abs(gps.getTimestamp() - gps.getServerTimestamp()) > 1000 * 60 * 20)
                             continue;
-                        gpsStateProcess.process(gps);//状态处理
+                        //gpsStateProcess.process(gps);//状态处理
                         stationInsideProcess.process(gps);//场站内外判定
                         reverseRouteProcess.process(gps);//反向路由处理
                         abnormalStateProcess.process(gps);//超速越界
@@ -134,20 +132,14 @@ public class DataHandleProcess {
                         outStationProcess.process(gps);//出站
-                        //logger.info("put start --" + Thread.currentThread().getName() + " -i: " + i);
                         GpsCacheData.putGps(gps);//历史gps缓存
-                        //logger.info("put end --" + Thread.currentThread().getName() + " -i: " + i);
                     } catch (Throwable e) {
                         logger.error("", e);
                     }
                 }
-
-                //logger.info("for end --" + Thread.currentThread().getName() + " -list size: " + list.size());
-
             } finally {
                 if (count != null)
                     count.countDown();
-                //logger.info(Thread.currentThread().getName() + " -countDown : " + count.getCount());
             }
         }
     }
@@ -164,19 +164,23 @@ public class InStationProcess {
         //将gps转换成下一个班次走向的站内信号（应对只有一个站内信号 即 发出时）
         transformUpDown(gps, next);
-        //下发调度指令
-        DirectivePushQueue.put6002(next, doneSum, "到站@系统");
-
         //套跑  -下发线路切换指令
         if (null != next && !next.getXlBm().equals(sch.getXlBm())) {
             gpsStatusManager.changeLine(next.getClZbh(), next.getXlBm(), "套跑@系统");
         }
+        /**
+         * 注意：
+         * 要先发运营指令，再发调度指令
+         */
+        if(null == next && gps.isService()){
+            nonService(sch, "结束@系统");//营运结束
+        }
+        //下发运营指令
+        DirectivePushQueue.put6003(next, "到站@系统");
-        if (null == next && gps.isService())
-            nonService(sch, "结束@系统");//班次结束
-        else if (null != next && dayOfSchedule.emptyService(next))
-            nonService(sch, "空驶@系统");//下一班非营运
+        //下发调度指令
+        DirectivePushQueue.put6002(next, doneSum, "到站@系统", "");
     }
     /**
@@ -123,7 +123,7 @@ public class RfidSignalHandle {
                 logger.info("RFID; 车辆：" + nbbm + " 班次：" + sch.getDfsj() + "到终点， 时间：" + sch.getZdsjActual());
                 //下发调度指令
-                DirectivePushQueue.put6002(next, doneSum, "rfid@系统");
+                DirectivePushQueue.put6002(next, doneSum, "rfid@系统", "");
                 //套跑  -下发线路切换指令
                 if(null != next && !next.getXlBm().equals(sch.getXlBm())){
@@ -11,6 +11,7 @@ import org.springframework.stereotype.Component;
 import java.util.Collection;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 /**
  * 设备营运状态/上下行 处理
@@ -19,7 +20,7 @@ import java.util.concurrent.ConcurrentHashMap;
 @Component
 public class ServiceStateHandle {
-    private static ConcurrentHashMap<String, StateBean> map;
+    private static ConcurrentMap<String, StateBean> map;
     @Autowired
     DirectiveService directiveService;
 package com.bsth.data.msg_queue;
+import com.bsth.data.schedule.DayOfSchedule;
 import com.bsth.entity.realcontrol.ScheduleRealInfo;
 import com.bsth.service.directive.DirectiveService;
 import org.slf4j.Logger;
@@ -9,7 +10,9 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Component;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
 /**
  * 到网关的指令推送队列 （系统发送的队列， 用户手动发送的不走这里）
@@ -22,13 +25,23 @@ public class DirectivePushQueue implements ApplicationContextAware {
     static DataPushThread thread;
     static DirectiveService directiveService;
     static long t;
-    static final int IDLE_TIME = 1000 * 30;
+
+    /**
+     * 下发运营指令6003的最小间隔时间
+     */
+    static final int MIN_SEND6003_SPACE = 1000 * 60;
+
+    /**
+     * 车辆 ——> 上次下发6003的时间
+     */
+    static ConcurrentMap<String, Long> lastSend6003Map;
     static {
         linkedList = new ConcurrentLinkedQueue<>();
+        lastSend6003Map = new ConcurrentHashMap<>();
     }
-    public static void put6002(ScheduleRealInfo sch, int finish, String sender){
+    public static void put6002(ScheduleRealInfo sch, int finish, String sender, String txtPrefix){
         if(null == sch)
             return;
         QueueData_Directive qd6002 = new QueueData_Directive();
@@ -36,21 +49,34 @@ public class DirectivePushQueue implements ApplicationContextAware {
         qd6002.setFinish(finish);
         qd6002.setSender(sender);
         qd6002.setCode("60_02");
+        qd6002.setTxtPrefix(txtPrefix);
         linkedList.add(qd6002);
     }
     public static void put6003(String nbbm, int state, int upDown, String sender){
+        long t = System.currentTimeMillis();
+        if(lastSend6003Map.containsKey(nbbm)
+                && t - lastSend6003Map.get(nbbm) < MIN_SEND6003_SPACE)
+            return; //最短下发间隔
+
         QueueData_Directive qd6003 = new QueueData_Directive();
         qd6003.setNbbm(nbbm);
         qd6003.setState(state);
         qd6003.setUpDown(upDown);
-        //qd6003.setSch(sch);
         qd6003.setSender(sender);
-
         qd6003.setCode("60_03");
         linkedList.add(qd6003);
+        lastSend6003Map.put(nbbm, t);
+    }
+
+    public static void put6003(ScheduleRealInfo sch, String sender){
+        int state = 0;//营运状态
+        if(DayOfSchedule.emptyService(sch))
+            state = 1;
+
+        put6003(sch.getClZbh(), state, Integer.parseInt(sch.getXlDir()), sender);
     }
     public static void put64(String nbbm, String lineCode, String sender){
@@ -64,10 +90,6 @@ public class DirectivePushQueue implements ApplicationContextAware {
         linkedList.add(qd64);
     }
-    public static boolean isIdle(){
-        return System.currentTimeMillis() - t > IDLE_TIME;
-    }
-
     public static void start(){
         if(thread != null){
             thread.interrupt();
@@ -108,7 +130,7 @@ public class DirectivePushQueue implements ApplicationContextAware {
                         code = qd.getCode();
                         if(code.equals("60_02")){
-                            directiveService.send60Dispatch(qd.getSch(), qd.getFinish(), qd.getSender());
+                            directiveService.send60Dispatch(qd.getSch(), qd.getFinish(), qd.getSender(), qd.getTxtPrefix());
                             log.info("directive 60_02 sch id: " + qd.getSch().getId());
                         }
                         else if(code.equals("60_03")){
@@ -29,6 +29,8 @@ public class QueueData_Directive {
     private String sender;
+    private String txtPrefix;
+
     public ScheduleRealInfo getSch() {
         return sch;
@@ -93,4 +95,12 @@ public class QueueData_Directive {
     public void setLineCode(String lineCode) {
         this.lineCode = lineCode;
     }
+
+    public String getTxtPrefix() {
+        return txtPrefix;
+    }
+
+    public void setTxtPrefix(String txtPrefix) {
+        this.txtPrefix = txtPrefix;
+    }
 }
@@ -93,7 +93,7 @@ public class PilotReport {
                         if (outSch.getJhlc() == null)
                             outSch = dayOfSchedule.next(outSch);
                         //下发调度指令
-                        DirectivePushQueue.put6002(outSch, dayOfSchedule.doneSum(nbbm), "请出@系统");
+                        DirectivePushQueue.put6002(outSch, dayOfSchedule.doneSum(nbbm), "请出@系统", "同意出场;");
                     }
                     break;
             }
@@ -904,6 +904,7 @@ public class DayOfSchedule {
     public List<ScheduleRealInfo> changeCar(ScheduleRealInfo sch, String newClZbh) {
         List<ScheduleRealInfo> ups = new ArrayList<>();
+        String oldClZbh = sch.getClZbh();
         //变更相关映射信息
         nbbmScheduleMap.remove(sch.getClZbh(), sch);
@@ -914,7 +915,7 @@ public class DayOfSchedule {
         //重新计算车辆当前执行班次
         reCalcExecPlan(newClZbh);
-        reCalcExecPlan(sch.getClZbh());
+        reCalcExecPlan(oldClZbh);
         //重新分组计划用车
         reCalcLineNbbms();
         return ups;
@@ -935,6 +936,8 @@ public class DayOfSchedule {
         ScheduleRealInfo sch = schAttrCalculator.calcCurrentExecSch(list);
         if(null != sch)
             carExecutePlanMap.put(nbbm, sch);
+        else
+            carExecutePlanMap.remove(nbbm);
     }
     /**
@@ -32,7 +32,7 @@ public interface DirectiveService extends BaseService&lt;D60, Integer&gt;{
 	 * @param @param finish  已完成的班次数
 	 * @throws
 	 */
-	int send60Dispatch(ScheduleRealInfo sch, int finish, String sender);
+	int send60Dispatch(ScheduleRealInfo sch, int finish, String sender,String txtPrefix);
 	/**
 	 * 
@@ -26,6 +26,7 @@ import com.bsth.websocket.handler.RealControlSocketHandler;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Splitter;
+import org.apache.commons.lang3.StringEscapeUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
@@ -78,6 +79,8 @@ public class DirectiveServiceImpl extends BaseServiceImpl&lt;D60, Integer&gt; implemen
     public int send60Phrase(String nbbm, String text, String sender) {
         D60 d60 = null;
         try {
+            text = StringEscapeUtils.unescapeHtml4(text);
+            text = text.replaceAll("#", "");
             d60 = create60Data(nbbm, text, (short) 0x00, null);
         } catch (Exception e) {
             logger.error("发送消息短语出现异常", e);
@@ -105,7 +108,7 @@ public class DirectiveServiceImpl extends BaseServiceImpl&lt;D60, Integer&gt; implemen
     }
     @Override
-    public int send60Dispatch(ScheduleRealInfo sch, int finish, String sender) {
+    public int send60Dispatch(ScheduleRealInfo sch, int finish, String sender, String txtPrefix) {
         D60 d60 = null;
         try {
             if (sch.isDestroy()) {
@@ -129,8 +132,15 @@ public class DirectiveServiceImpl extends BaseServiceImpl&lt;D60, Integer&gt; implemen
             long t = System.currentTimeMillis() + 1000 * 30,
                     alarmTime = sch.getDfsjT() < t?t:sch.getDfsjT();
+            if(StringUtils.isNotEmpty(txtPrefix)){
+                text = txtPrefix + text;
+            }
+            text = StringEscapeUtils.unescapeHtml4(text);
+            int state = 0;//营运状态
+            if(dayOfSchedule.emptyService(sch))
+                state = 1;
             d60 = new DirectiveCreator().createD60_01(sch.getClZbh(), text, Integer.parseInt(sch.getXlDir())
-                    , 0, new Date(alarmTime));
+                    , state, new Date(alarmTime));
             d60.setLineCode(sch.getXlBm());
         } catch (Exception e) {
@@ -170,6 +180,7 @@ public class DirectiveServiceImpl extends BaseServiceImpl&lt;D60, Integer&gt; implemen
             dayOfDirectives.put60(d60, false);
             d60Repository.save(d60);
         }
+
         return code;
     }
@@ -196,7 +207,7 @@ public class DirectiveServiceImpl extends BaseServiceImpl&lt;D60, Integer&gt; implemen
         ScheduleRealInfo sch = dayOfSchedule.get(id);
         // 车辆已完成班次
         int finish = dayOfSchedule.doneSum(sch.getClZbh());
-        return send60Dispatch(sch, finish, sender);
+        return send60Dispatch(sch, finish, sender, "");
     }
     @Override
@@ -17,7 +17,7 @@
                     <select name="dType">
                         <option value="-1">全部</option>
                         <option value="3">消息短语</option>
-                        <option value="0">调度指令</option>
+                        <option value="0" selected>调度指令</option>
                         <option value="1">运营指令</option>
                         <option value="2">线路切换指令</option>
                     </select>