按秒计算排队长度时，用停车状态为停止的车辆数来计算排队长度

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalDirectionSecondMetricFunction.java

 package com.zhht.irn.functions;
 
-import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
 import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
@@ -66,7 +65,7 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
                 });
 
 
-                // 计算两个指标，一个 “发生到达，未进入的车辆数”， 一个“发生到达数”，粒度计算到车道级
+                // 计算一个“发生到达数”，粒度计算到车道级
                 for(Map.Entry<String, List<TravelEvent>> entry : lineMap.entrySet()) {
                     List<TravelEvent> travelEventList = entry.getValue();
 
@@ -89,11 +88,9 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
                         if(e.getEventType() == EventType.STOP) {
                             recordStopStatusMap.put(e.getRecordId(), true);
                             globalRecordStopStatusMap.put(e.getRecordId(), true);
-                            System.out.println("you che 停了" + e.getRecordId());
                         } else if(e.getEventType() == EventType.STARTUP) {
                             recordStopStatusMap.put(e.getRecordId(), false);
                             globalRecordStopStatusMap.put(e.getRecordId(), false);
-                            System.out.println("you che 启动了" + e.getRecordId());
                         }
                     }
 
@@ -106,22 +103,13 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
                         }
                     }).collect(Collectors.toList());
 
-                    List<TravelEvent> lineArrivedAndInCrossList = travelEventList.stream().filter(record -> {
-                        return record.getEventType() == EventType.INCROSS && recordIdSet.contains(record.getRecordId());
-                    }).collect(Collectors.toList());
-
                     Set<String> arrivedRecordIdSet = new HashSet<>();
-                    Set<String> arrivedAndInCrossRecordIdSet = new HashSet<>();
 
                     for(TravelEvent travel:  lineArrivedList) {
                         arrivedRecordIdSet.add(travel.getRecordId());
                     }
-                    for(TravelEvent travel: lineArrivedAndInCrossList) {
-                        arrivedAndInCrossRecordIdSet.add(travel.getRecordId());
-                    }
 
                     int lineArrivedCount = arrivedRecordIdSet.size();
-                    int lineArrivedAndInCrossCount = arrivedAndInCrossRecordIdSet.size();
 
                     int stopFlagCount = 0;
                     int stopCount = 0;
@@ -138,7 +126,7 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
                     metric.setCurrentSecond(start);
                     metric.setCycleOrder(cycle.getCycleOrder());
                     metric.setAccArrivedFlow(lineArrivedCount);
-                    metric.setAccQueueLength(lineArrivedCount - lineArrivedAndInCrossCount);
+                    metric.setAccQueueLength(stopFlagCount);
                     metric.setStopFlagCount(stopFlagCount);
                     metric.setAccStopCount(stopCount);
                     collector.collect(metric);
@@ -149,7 +137,6 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
         // 将本周期结束后还处于停车状态的车辆放到listState中
         listState.clear();
         for(Map.Entry<String, Boolean> en : globalRecordStopStatusMap.entrySet()) {
-            System.out.println("当前周期的不同记录数：" + globalRecordStopStatusMap.size());
             if(en.getValue() == true) {
                 System.out.println("stay stop status car record id:" + en.getKey());
                 listState.add(en.getKey());

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalLaneMetricProcessFunction2.java

-package com.zhht.irn.functions;
-
-import com.zhht.irn.entity.Cycle;
-import com.zhht.irn.entity.TravelEvent;
-import org.apache.flink.api.common.state.MapState;
-import org.apache.flink.api.common.state.MapStateDescriptor;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
-import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
-import org.apache.flink.util.Collector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * 计算车道级指标的函数
- *
- */
-public class CalLaneMetricProcessFunction2 extends KeyedCoProcessFunction<String, Cycle, TravelEvent, Object> {
-    private static final Logger logger = LoggerFactory.getLogger(CalLaneMetricProcessFunction2.class);
-    // state 管理当前周期中存储的旅行事件
-    private MapState<String, List<TravelEvent>> cycleTravel;
-
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        cycleTravel = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravel", String.class, Object.class));
-    }
-
-    @Override
-    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, Object>.Context ctx, Collector<Object> out) throws Exception {
-        logger.info("进来个周期数据了：" + cycle.toString());
-        // 在此进行指标计算
-    }
-
-    @Override
-    public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, Object>.Context ctx, Collector<Object> out) throws Exception {
-        logger.info("进来个事件数据了：" + event.toString());
-        if(cycleTravel.get(event.getCrossId()) == null) {
-            cycleTravel.put(event.getCrossId(), new ArrayList<>());
-        }
-        cycleTravel.get(event.getCrossId()).add(event);
-
-    }
-}

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelMetricSink.java

@@ -37,6 +37,7 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     }
     @Override
     public void close() throws Exception {
+        System.out.println("执行close方法了");
         super.close();
         if(insertPstmt != null) insertPstmt.close();
         if(connection != null) connection.close();