修改按方向的任务的计算策略，改为connected stream来计算

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

@@ -10,7 +10,7 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
 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.co.ProcessJoinFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
 import org.apache.flink.util.Collector;
 
 import java.util.*;
@@ -24,7 +24,7 @@ import java.util.stream.Collectors;
  * @author 吴延飞
  * @date 2022-11-15 11:23:00
  */
-public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle, TravelEvent, DirectionEvalSecondMetric> {
+public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, DirectionEvalSecondMetric> {
 
     private MapState<String, List<TravelEvent>> mapState;
 
@@ -38,27 +38,15 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
     }
 
     @Override
-    public void processElement(Cycle cycle, TravelEvent travelEvent, ProcessJoinFunction<Cycle, TravelEvent, DirectionEvalSecondMetric>.Context ctx, Collector<DirectionEvalSecondMetric> out) throws Exception {
-        // 进入该方法表明，周期已经到了，而且周期已经向过车记录interval join了[-9,1]分钟的旅行事件数据
-
-        // 先累计旅行事件数据到 mapState 中, 只累计到达和进入事件，其他类型计算用不上
-        List<TravelEvent> cachedTravelEvent = mapState.get(cycle.getCrossCode());
-        if(cachedTravelEvent == null) {
-            cachedTravelEvent = new ArrayList<>();
-            mapState.put(cycle.getCrossCode(), cachedTravelEvent);
-        }
-        if(travelEvent.getEventType() == EventType.ARRIVED || travelEvent.getEventType() == EventType.INCROSS
-            || travelEvent.getEventType() == EventType.STOP || travelEvent.getEventType() == EventType.STARTUP)
-        cachedTravelEvent.add(travelEvent);
-
-        // 确定触发条件
-        if(Long.parseLong(travelEvent.getEventTime()) - Constant.CYCLE_CALCULATE_LATENESS > Long.parseLong(cycle.getEndDateTime())) {
+    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, DirectionEvalSecondMetric>.Context context, Collector<DirectionEvalSecondMetric> collector) throws Exception {
+        // 周期数据进入，开始计算
         long cycleStart = Long.parseLong(cycle.getBeginDateTime());
         long cycleEnd = Long.parseLong(cycle.getEndDateTime());
-
+        List<TravelEvent> cachedTravelEvent = mapState.get(cycle.getCrossCode());
+        Map<String, Boolean> globalRecordStopStatusMap = new HashMap<>();
         // 计算每个车道的到达流量, 按秒计算
         for(long start = cycleStart; start <= cycleEnd; start += 1000) {
-                // 首先过滤出到当前秒的事件数据
+            // 首先过滤出到当前周期的事件数据
             long finalStart = start;
             if (cachedTravelEvent != null) {
                 List<TravelEvent> currentEventList = cachedTravelEvent.stream().filter(event -> {
@@ -92,14 +80,20 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                     for(String recordId : listState.get()) {
                         recordStopStatusMap.put(recordId, true);
                     }
+                    // 严格一点，这里要对事件进行时间的排序, 因为乱序会导致车辆的状态不正确
+                    Collections.sort(travelEventList, Comparator.comparing(TravelEvent::getEventTime));
+
                     // 再处理本周期内发生的事件
                     for(int i = 0; i < travelEventList.size(); i ++) {
-                            // 严格一点，这里要对事件进行时间的排序，暂且不做
                         TravelEvent e = travelEventList.get(i);
                         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());
                         }
                     }
 
@@ -112,8 +106,6 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                         }
                     }).collect(Collectors.toList());
 
-
-
                     List<TravelEvent> lineArrivedAndInCrossList = travelEventList.stream().filter(record -> {
                         return record.getEventType() == EventType.INCROSS && recordIdSet.contains(record.getRecordId());
                     }).collect(Collectors.toList());
@@ -138,7 +130,6 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                         stopCount ++;
                     }
 
-
                     DirectionEvalSecondMetric metric = new DirectionEvalSecondMetric();
                     metric.setCycleStart(cycleStart);
                     metric.setCycleEnd(cycleEnd);
@@ -150,14 +141,43 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                     metric.setAccQueueLength(lineArrivedCount - lineArrivedAndInCrossCount);
                     metric.setStopFlagCount(stopFlagCount);
                     metric.setAccStopCount(stopCount);
+                    collector.collect(metric);
+                }
+            }
+        }
 
-                        out.collect(metric);
+        // 将本周期结束后还处于停车状态的车辆放到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());
             }
         }
+
+        if(cachedTravelEvent != null) {
+            List<TravelEvent> newOne = new ArrayList<>();
+            for(TravelEvent t : cachedTravelEvent) {
+                // 删除掉这个周期结束时间之间的数据
+                if(Long.parseLong(t.getEventTime()) >= Long.parseLong(cycle.getEndDateTime())) {
+                    newOne.add(t);
+                }
+            }
+            cachedTravelEvent = newOne;
+            mapState.put(cycle.getCrossCode(), cachedTravelEvent);
+        }
     }
 
-            // 计算后要清空
-            cachedTravelEvent.clear();
+    @Override
+    public void processElement2(TravelEvent travelEvent, KeyedCoProcessFunction<String, Cycle, TravelEvent, DirectionEvalSecondMetric>.Context context, Collector<DirectionEvalSecondMetric> collector) throws Exception {
+        List<TravelEvent> cachedTravelEvent = mapState.get(travelEvent.getCrossId());
+        if(cachedTravelEvent == null) {
+            cachedTravelEvent = new ArrayList<>();
+            mapState.put(travelEvent.getCrossId(), cachedTravelEvent);
         }
+        if(travelEvent.getEventType() == EventType.ARRIVED || travelEvent.getEventType() == EventType.INCROSS
+                || travelEvent.getEventType() == EventType.STOP || travelEvent.getEventType() == EventType.STARTUP)
+            cachedTravelEvent.add(travelEvent);
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/DirectionEvalJob.java

@@ -4,15 +4,20 @@ import com.alibaba.fastjson.JSON;
 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;
 import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
 import com.zhht.irn.sink.DirectionEvalSecondSink;
+import com.zhht.irn.source.CycleMockSource;
+import com.zhht.irn.source.TravelEventMockSource;
 import com.zhht.irn.utils.FlinkUtils;
 import com.zhht.irn.utils.StringUtils;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.KeyedStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
 import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,6 +43,7 @@ public class DirectionEvalJob {
         DataStream<Cycle> cycleStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
+                    System.out.println(record);
                     return JSON.parseObject(record, Cycle.class);
                 });
 
@@ -45,6 +51,7 @@ public class DirectionEvalJob {
         DataStream<TravelEvent> travelEventStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
+                    System.out.println(record);
                     return JSON.parseObject(record, TravelEvent.class);
                 }).filter(event -> {
                     return StringUtils.isNotEmpty(event.getEventLineId());
@@ -67,8 +74,7 @@ public class DirectionEvalJob {
         // 周期数据join 旅行事件
         cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
                 .keyBy(record -> record.getCrossCode())
-                .intervalJoin(travelEventKeyedStream)
-                .between(Time.minutes(-9), Time.minutes(1))
+                .connect(travelEventKeyedStream)
                 .process(new CalDirectionSecondMetricFunction())
                 .addSink(new DirectionEvalSecondSink()).setParallelism(4);
 

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

@@ -43,7 +43,6 @@ public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecon
 
     @Override
     public void invoke(DirectionEvalSecondMetric metric, Context context) throws Exception {
-        System.out.println("===== insert metric {" + metric + "}");
         insertPstmt.setString(1, metric.getCrossId());
         insertPstmt.setString(2, metric.getLineId());
         insertPstmt.setLong(3, metric.getCycleOrder());