自定义时间窗口函数完成和测试完成

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java

@@ -16,9 +16,9 @@ public class Cycle {
     //信控机编号
     private String signalCode;
     // 周期开始时间
-    private String beginDateTime;
+    private Date beginDateTime;
     // 周期结束时间
-    private String endDateTime;
+    private Date endDateTime;
     // 开始模式编码
     private String beginControlModelCode;
     // 结束模式编码
@@ -54,19 +54,19 @@ public class Cycle {
         this.crossCode = crossCode;
     }
 
-    public String getBeginDateTime() {
+    public Date getBeginDateTime() {
         return beginDateTime;
     }
 
-    public void setBeginDateTime(String beginDateTime) {
+    public void setBeginDateTime(Date beginDateTime) {
         this.beginDateTime = beginDateTime;
     }
 
-    public String getEndDateTime() {
+    public Date getEndDateTime() {
         return endDateTime;
     }
 
-    public void setEndDateTime(String endDateTime) {
+    public void setEndDateTime(Date endDateTime) {
         this.endDateTime = endDateTime;
     }
 
@@ -149,4 +149,24 @@ public class Cycle {
     public void setEndPlanId(Integer endPlanId) {
         this.endPlanId = endPlanId;
     }
+
+    @Override
+    public String toString() {
+        return "Cycle{" +
+                "id=" + id +
+                ", crossCode='" + crossCode + '\'' +
+                ", signalCode='" + signalCode + '\'' +
+                ", beginDateTime=" + beginDateTime +
+                ", endDateTime=" + endDateTime +
+                ", beginControlModelCode='" + beginControlModelCode + '\'' +
+                ", endControlModelCode='" + endControlModelCode + '\'' +
+                ", beginPlanId=" + beginPlanId +
+                ", endPlanId=" + endPlanId +
+                ", duration=" + duration +
+                ", cycleOrder=" + cycleOrder +
+                ", greenRatio=" + greenRatio +
+                ", lossTime=" + lossTime +
+                ", stageList=" + stageList +
+                '}';
+    }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/TravelEvent.java

@@ -39,4 +39,91 @@ public class TravelEvent {
 
     // 备注
     private String remark;
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public String getCrossId() {
+        return crossId;
+    }
+
+    public void setCrossId(String crossId) {
+        this.crossId = crossId;
+    }
+
+    public String getRecordId() {
+        return recordId;
+    }
+
+    public void setRecordId(String recordId) {
+        this.recordId = recordId;
+    }
+
+    public Date getRecordTime() {
+        return recordTime;
+    }
+
+    public void setRecordTime(Date recordTime) {
+        this.recordTime = recordTime;
+    }
+
+    public Long getEventLineId() {
+        return eventLineId;
+    }
+
+    public void setEventLineId(Long eventLineId) {
+        this.eventLineId = eventLineId;
+    }
+
+    public Double getEventSpeed() {
+        return eventSpeed;
+    }
+
+    public void setEventSpeed(Double eventSpeed) {
+        this.eventSpeed = eventSpeed;
+    }
+
+    public Date getEventTime() {
+        return eventTime;
+    }
+
+    public void setEventTime(Date eventTime) {
+        this.eventTime = eventTime;
+    }
+
+    public EventType getEventType() {
+        return eventType;
+    }
+
+    public void setEventType(EventType eventType) {
+        this.eventType = eventType;
+    }
+
+    public String getRemark() {
+        return remark;
+    }
+
+    public void setRemark(String remark) {
+        this.remark = remark;
+    }
+
+    @Override
+    public String toString() {
+        return "TravelEvent{" +
+                "id=" + id +
+                ", crossId='" + crossId + '\'' +
+                ", recordId='" + recordId + '\'' +
+                ", recordTime=" + recordTime +
+                ", eventLineId=" + eventLineId +
+                ", eventSpeed=" + eventSpeed +
+                ", eventTime=" + eventTime +
+                ", eventType=" + eventType +
+                ", remark='" + remark + '\'' +
+                '}';
+    }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/LaneMetricJob.java

+package com.zhht.irn.flow;
+
+import com.alibaba.fastjson.JSON;
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.TravelEvent;
+import com.zhht.irn.functions.CalLaneMetricProcessFunction;
+import com.zhht.irn.source.CycleMockSource;
+import com.zhht.irn.source.TravelEventMockSource;
+import com.zhht.irn.trigger.CycleTimeTrigger;
+import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.window.CycleWindowAssigner;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.streaming.api.windowing.windows.Window;
+import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.*;
+
+/**
+ * 车道级指标的计算任务
+ * 按照周期进行实时计算车道级的指标
+ * 计算的指标有：
+ *  最大排队长度、有效绿灯时间、通行能力、交通流量、饱和度、剩余承载力、识别区空间占有率
+ *
+ * @author 吴延飞
+ * @date 2022-11-13 13:55:00
+ */
+public class LaneMetricJob {
+    private static final Logger logger = LoggerFactory.getLogger(LaneMetricJob.class);
+
+    public static void main(String[] args) throws Exception {
+        // 创建实时数据流上下文
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        // 旅行事件数据流的 watermark 生成逻辑
+        WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30))
+                .withTimestampAssigner((travelEvent, timestamp) -> travelEvent.getEventTime().getTime());
+
+        // 周期结束数据流的 cycle 生成逻辑
+        WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness(Duration.ofSeconds(10))
+                .withTimestampAssigner((cycle, timeStamp) -> cycle.getBeginDateTime().getTime());
+
+        // 获取实时旅行事件数据流
+//        DataStream<TravelEvent> eventStream = FlinkUtils.createKafkaStream(args, env)
+        DataStream<TravelEvent> eventStream = env.addSource(new TravelEventMockSource())
+                // 格式解析，将Sting转化为实体类, 存在无法直接从String转为Entity的情况，需字段单独解析 @TODO
+                .map((record) -> {
+                    System.out.println(record);
+                    TravelEvent travelEvent = JSON.parseObject(record, TravelEvent.class);
+                    System.out.println(travelEvent);
+                    return travelEvent;
+                })
+                // 给数据流添加水印
+                .assignTimestampsAndWatermarks(travelEventWatermarkStrategy);
+
+        // 获取实时周期结束数据流
+//        DataStream<Cycle> cycleStream = FlinkUtils.createKafkaStream(args, env)
+        DataStream<Cycle> cycleStream = env.addSource(new CycleMockSource())
+                // 格式解析，将Sting转化为实体类, 存在无法直接从String转为Entity的情况，需字段单独解析 @TODO
+                .map((record) ->  {
+                    System.out.println(record);
+                    Cycle cycle = JSON.parseObject(record, Cycle.class);
+                    System.out.println(cycle);
+                    return cycle;
+                })
+                // 给数据流添加水印
+                .assignTimestampsAndWatermarks(cycleWatermarkStrategy);
+
+        // 通过join操作，获取一个Tuple<List<TravelEvent>, Cycle> 类型的数据流
+        DataStream<Tuple2<Cycle, TravelEvent>> joinStream = cycleStream.coGroup(eventStream)
+                // 根据路口分组
+                .where(cycle  -> cycle.getCrossCode())
+                .equalTo(travelEvent -> travelEvent.getCrossId())
+                // 根据周期来进行窗口划分
+                .window(new CycleWindowAssigner())
+                .trigger(new CycleTimeTrigger())
+                //
+                .apply(new CoGroupFunction<Cycle, TravelEvent, Tuple2<Cycle, TravelEvent>>() {
+                    @Override
+                    public void coGroup(Iterable<Cycle> iterable, Iterable<TravelEvent> iterable1, Collector<Tuple2<Cycle, TravelEvent>> collector) throws Exception {
+                        System.out.println(1);
+                    }
+                });
+
+        joinStream.keyBy(value -> value.f0.getCrossCode())
+                .process(new CalLaneMetricProcessFunction())
+                .print();
+
+        env.execute();
+
+    }
+}

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

+package com.zhht.irn.functions;
+
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.TravelEvent;
+import com.zhht.irn.utils.MySQLUtils;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.util.Collector;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 计算车道级指标的函数
+ *
+ */
+public class CalLaneMetricProcessFunction extends KeyedProcessFunction<String, Tuple2<Cycle, TravelEvent>, Object> {
+    // 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 processElement(Tuple2<Cycle,TravelEvent> value, KeyedProcessFunction<String, Tuple2<Cycle, TravelEvent>, Object>.Context ctx, Collector<Object> out) throws Exception {
+        if(value.f0 == null) {
+            // 表示周期数据还没到，直接缓存到mapState中
+            String key = value.f0.getCrossCode(); // 以路口为key进行存储
+            if(cycleTravel.get(key) != null) {
+                cycleTravel.get(key).add(value.f1);
+            }else {
+                List<TravelEvent> list = new ArrayList<>();
+                list.add(value.f1);
+                cycleTravel.put(key, list);
+            }
+        } else {
+            // 代表周期已到，需要进行相应的指标计算了
+
+
+        }
+    }
+}

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

@@ -52,8 +52,8 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
         insertCyclePstmt.setLong(1, cycle.getId());
         insertCyclePstmt.setString(2, cycle.getCrossCode());
         insertCyclePstmt.setString(3, cycle.getSignalCode());
-        insertCyclePstmt.setString(4, cycle.getBeginDateTime());
-        insertCyclePstmt.setString(5, cycle.getEndDateTime());
+        insertCyclePstmt.setString(4, DateUtils.dateToString(cycle.getBeginDateTime(), "yyyy-MM-DD HH:mm:ss"));
+        insertCyclePstmt.setString(5, DateUtils.dateToString(cycle.getEndDateTime(), "yyyy-MM-DD HH:mm:ss"));
         insertCyclePstmt.setInt(6,cycle.getDuration());
         insertCyclePstmt.setInt(7,cycle.getCycleOrder());
         insertCyclePstmt.setDouble(8,cycle.getGreenRatio());
@@ -66,8 +66,8 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
             insertPhasePstmt.setLong(1,cycle.getCycleOrder());
             insertPhasePstmt.setString(2,phase.getCrossCode());
             insertPhasePstmt.setString(3,phase.getPhaseValue());
-            insertPhasePstmt.setString(4, cycle.getBeginDateTime());
-            insertPhasePstmt.setString(5, cycle.getEndDateTime());
+            insertCyclePstmt.setString(4, DateUtils.dateToString(cycle.getBeginDateTime(), "yyyy-MM-DD HH:mm:ss"));
+            insertCyclePstmt.setString(5, DateUtils.dateToString(cycle.getEndDateTime(), "yyyy-MM-DD HH:mm:ss"));
             insertPhasePstmt.setInt(6,cycle.getDuration());
             insertPhasePstmt.setInt(7,phase.getPhaseOrder());
             insertPhasePstmt.setDouble(8,phase.getGreenRatio());

realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CirlceSource.java

-package com.zhht.irn.source;
-
-import com.alibaba.fastjson.JSON;
-import com.zhht.irn.entity.Cycle;
-import com.zhht.irn.entity.Stage;
-import com.zhht.irn.utils.DateUtils;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.util.TimeUtils;
-
-import java.sql.Timestamp;
-import java.util.*;
-
-/**
- * @Description:
- * @Author ：Marinh
- * @Param:
- * @retrun:
- * @Creat :2022-11-01-14:40
- **/
-public class CirlceSource implements SourceFunction<Cycle> {
-    private static boolean isRunning = true;
-    private static int i = 0;
-
-    @Override
-    public void run(SourceFunction.SourceContext<Cycle> sourceContext) throws Exception {
-        while (isRunning){
-            Calendar beforeTime = Calendar.getInstance();
-            Date date = new Date();
-            sourceContext.collect(getCycle(beforeTime,date));
-            //每个3秒发送一条数据
-            Thread.sleep(60000);
-        }
-    }
-
-    public static Cycle getCycle(Calendar beforeTime,Date date){
-        Cycle cycle = new Cycle();
-        cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
-        cycle.setCrossCode("13070201");
-        cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
-        cycle.setEndDateTime(DateUtils.formatTime(date));
-        cycle.setDuration(120);
-        cycle.setCycleOrder(1);
-        ArrayList<Stage> phases = new ArrayList<>();
-        for(int i=0;i<3;i++){
-            Stage phase = new Stage();
-            phase.setPhaseValue("10"+i);
-            if(i==0){
-                phase.setPhaseValue("1");
-                phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,120));
-                phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
-                phase.setDuration(30);
-                phase.setGreen(30);
-                phase.setAllRed(87);
-                phase.setGreen(3);
-            }
-            if(i==1){
-                phase.setPhaseValue("2");
-                phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
-                phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
-                phase.setDuration(30);
-                phase.setGreen(30);
-                phase.setAllRed(87);
-                phase.setGreen(3);
-            }
-            if(i==2){
-                phase.setPhaseValue("3");
-                phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
-                phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,0));
-                phase.setDuration(60);
-                phase.setGreen(60);
-                phase.setAllRed(57);
-                phase.setGreen(3);
-            }
-            phases.add(phase);
-        }
-        cycle.setStageList(phases);
-        return cycle;
-    }
-
-    @Override
-    public void cancel() {
-        isRunning=false;
-    }
-
-}
+//package com.zhht.irn.source;
+//
+//import com.alibaba.fastjson.JSON;
+//import com.zhht.irn.entity.Cycle;
+//import com.zhht.irn.entity.Stage;
+//import com.zhht.irn.utils.DateUtils;
+//import org.apache.flink.streaming.api.functions.source.SourceFunction;
+//import org.apache.flink.util.TimeUtils;
+//
+//import java.sql.Timestamp;
+//import java.util.*;
+//
+///**
+// * @Description:
+// * @Author ：Marinh
+// * @Param:
+// * @retrun:
+// * @Creat :2022-11-01-14:40
+// **/
+//public class CirlceSource implements SourceFunction<Cycle> {
+//    private static boolean isRunning = true;
+//    private static int i = 0;
+//
+//    @Override
+//    public void run(SourceFunction.SourceContext<Cycle> sourceContext) throws Exception {
+//        while (isRunning){
+//            Calendar beforeTime = Calendar.getInstance();
+//            Date date = new Date();
+//            sourceContext.collect(getCycle(beforeTime,date));
+//            //每个3秒发送一条数据
+//            Thread.sleep(60000);
+//        }
+//    }
+//
+//    public static Cycle getCycle(Calendar beforeTime,Date date){
+//        Cycle cycle = new Cycle();
+//        cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
+//        cycle.setCrossCode("13070201");
+//        cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
+//        cycle.setEndDateTime(DateUtils.formatTime(date));
+//        cycle.setDuration(120);
+//        cycle.setCycleOrder(1);
+//        ArrayList<Stage> phases = new ArrayList<>();
+//        for(int i=0;i<3;i++){
+//            Stage phase = new Stage();
+//            phase.setPhaseValue("10"+i);
+//            if(i==0){
+//                phase.setPhaseValue("1");
+//                phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,120));
+//                phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
+//                phase.setDuration(30);
+//                phase.setGreen(30);
+//                phase.setAllRed(87);
+//                phase.setGreen(3);
+//            }
+//            if(i==1){
+//                phase.setPhaseValue("2");
+//                phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
+//                phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
+//                phase.setDuration(30);
+//                phase.setGreen(30);
+//                phase.setAllRed(87);
+//                phase.setGreen(3);
+//            }
+//            if(i==2){
+//                phase.setPhaseValue("3");
+//                phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
+//                phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,0));
+//                phase.setDuration(60);
+//                phase.setGreen(60);
+//                phase.setAllRed(57);
+//                phase.setGreen(3);
+//            }
+//            phases.add(phase);
+//        }
+//        cycle.setStageList(phases);
+//        return cycle;
+//    }
+//
+//    @Override
+//    public void cancel() {
+//        isRunning=false;
+//    }
+//
+//}

realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CycleMockSource.java

+package com.zhht.irn.source;
+
+import com.zhht.irn.utils.DateUtils;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+
+import java.util.Date;
+
+public class CycleMockSource extends RichSourceFunction<String> {
+    @Override
+    public void run(SourceContext<String> ctx) throws Exception {
+        // 每隔3、4、5秒发送一条TravelEvent事件
+        String cycleInfoJsonTemplate = "{\n" +
+                "\t\"id\": 1,\n" +
+                "\t\"crossCode\": \"137\",\n" +
+                "\t\"beginDateTime\": %s,\n" +
+                "\t\"endDateTime\": %s,\n" +
+                "  \"signalCode\": \"1234\",\n" +
+                "\t\"duration\": 90,\n" +
+                "\t\"cycleOrder\": 1,\n" +
+                "  \"beginControlModeCode\": \"12\",\n" +
+                "  \"endControlModeCode\": \"13\",\n" +
+                "  \"beginPlanId\": 43,\n" +
+                "  \"endPlanId\": 42,\n" +
+                "\t\"stageList\": [{\n" +
+                "\t\t\t\"phaseValue\": \"101\",\n" +
+                "\t\t\t\"beginDateTime\": %s,\n" +
+                "\t\t\t\"endDateTime\": %s,\n" +
+                "      \"beginControlModeCode\": \"12\",\n" +
+                "    \t\"endControlModeCode\": \"13\",\n" +
+                "    \t\"beginPlanId\": 43,\n" +
+                "    \t\"endPlanId\": 42,\n" +
+                "\t\t\t\"duration\": 60,\n" +
+                "\t\t\t\"green\": 57,\n" +
+                "\t\t\t\"yellow\": 3,\n" +
+                "\t\t\t\"allRed\": 0,\n" +
+                "\t\t\t\"phaseOrder\": 1\n" +
+                "\t\t},\n" +
+                "\t\t{\n" +
+                "\t\t\t\"phaseValue\": \"102\",\n" +
+                "\t\t\t\"beginDateTime\": %s,\n" +
+                "\t\t\t\"endDateTime\": %s,\n" +
+                "      \"beginControlModeCode\": \"12\",\n" +
+                "    \t\"endControlModeCode\": \"13\",\n" +
+                "    \t\"beginPlanId\": 43,\n" +
+                "    \t\"endPlanId\": 42,\n" +
+                "\t\t\t\"duration\": 30,\n" +
+                "\t\t\t\"green\": 27,\n" +
+                "\t\t\t\"yellow\": 3,\n" +
+                "\t\t\t\"allRed\": 0,\n" +
+                "\t\t\t\"phaseOrder\": 2\n" +
+                "\t\t}\n" +
+                "\t]\n" +
+                "}";
+            Thread.sleep(90 * 1000);
+            // 发送捕获数据
+            Date now = new Date();
+            long cycleEnd  = now.getTime();
+            long cycleStart = now.getTime() - 90 * 1000;
+            String start = DateUtils.dateToString(new Date(cycleStart), "yyyy-MM-dd HH:mm:ss");
+            String end = DateUtils.dateToString(new Date(cycleEnd), "yyyy-MM-dd HH:mm:ss");
+            ctx.collect(String.format(cycleInfoJsonTemplate, 1, "1000", start, end, start, end,start, end));
+
+    }
+
+    @Override
+    public void cancel() {
+
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/source/TravelEventMockSource.java

+package com.zhht.irn.source;
+
+import com.zhht.irn.utils.DateUtils;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+
+import java.util.Date;
+
+public class TravelEventMockSource extends RichSourceFunction<String> {
+    @Override
+    public void run(SourceContext<String> ctx) throws Exception {
+        // 每隔3、4、5秒发送一条TravelEvent事件
+        String travelInfoJsonTemplate = "{ \"id\": %d, \"crossId\": \"137\", \"recordId\": %s, \"recordTime\": %s," +
+                "\"eventLineId\": %s, \"eventTime\": %s, \"eventSpeed\": 36.4421, \"eventType\": \"%s\", \"remark\": \"\"}";
+            // 发送捕获数据
+            Date now = new Date();
+            ctx.collect(String.format(travelInfoJsonTemplate, 1, "1000", now.getTime(), "111", now.getTime(), "FIRST"));
+
+            Thread.sleep(3000);
+
+            now = new Date();
+            ctx.collect(String.format(travelInfoJsonTemplate, 2, "1000", now.getTime(), "111", now.getTime(), "ARRIVED"));
+
+            Thread.sleep(2000);
+            now = new Date();
+            ctx.collect(String.format(travelInfoJsonTemplate, 3, "1000", now.getTime(), "111", now.getTime(), "INTCROSS"));
+
+            Thread.sleep(5000);
+            now = new Date();
+            ctx.collect(String.format(travelInfoJsonTemplate, 4, "1000", now.getTime(), "111", now.getTime(), "OUTCROSS"));
+
+
+            Thread.sleep(2000);
+            now = new Date();
+            ctx.collect(String.format(travelInfoJsonTemplate, 5, "1000", now.getTime(), "111", now.getTime(), "AWAY"));
+
+            Thread.sleep(3000);
+            now = new Date();
+            ctx.collect(String.format(travelInfoJsonTemplate, 6, "1000", now.getTime(), "111", now.getTime(), "LAST"));
+    }
+
+    @Override
+    public void cancel() {
+
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/trigger/CycleTimeTrigger.java

+package com.zhht.irn.trigger;
+
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.TravelEvent;
+import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+public class CycleTimeTrigger extends Trigger<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
+    @Override
+    public TriggerResult onElement(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
+        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
+            // if the watermark is already past the window fire immediately
+            return TriggerResult.FIRE;
+        } else {
+            ctx.registerEventTimeTimer(window.maxTimestamp());
+            return TriggerResult.CONTINUE;
+        }
+    }
+
+    @Override
+    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
+        return TriggerResult.CONTINUE;
+    }
+
+    @Override
+    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
+        return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
+    }
+
+    @Override
+    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
+        ctx.deleteEventTimeTimer(window.maxTimestamp());
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java

@@ -11,7 +11,10 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
 import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Properties;
@@ -19,6 +22,7 @@ import java.util.concurrent.TimeUnit;
 
 
 public class FlinkUtils {
+    private static final Logger logger = LoggerFactory.getLogger(FlinkUtils.class);
 
     public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
@@ -89,13 +93,44 @@ public class FlinkUtils {
         return env.addSource(kafkaConsumer);
     }
 
-    // /Users/rocky/Desktop/flink/11/pk.properties
-    public static void main(String[] args) throws Exception{
-        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
-        // 参数是分成2大类：1)必填 2)选填
-        String groupId = tool.get("group.id", "test1");
-        String servers = tool.getRequired("bootstrap.servers");
-        System.out.println(groupId);
-        System.out.println(servers);
+
+    public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment) {
+        // 获取Kafka配置文件
+        ParameterTool finkArgs = ParameterTool.fromArgs(args);
+        String kafkaConfigPath = finkArgs.getRequired("kafka.conf");
+        ParameterTool kafkaTool;
+        try {
+            kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
+        } catch (IOException e) {
+            logger.error("kafka 配置文件未设置，请检查!");
+            throw new RuntimeException(e);
+        }
+
+        // 加载Kafka信息 { topic 相关 }
+        String groupId = kafkaTool.get("group.id", "test1");
+        String servers = kafkaTool.getRequired("bootstrap.servers");
+        List<String> topics = Arrays.asList(kafkaTool.getRequired("kafka.input.topics").split(","));
+        String autoCommit = kafkaTool.get("enable.auto.commit", "false");
+        String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
+
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", servers);
+        properties.setProperty("group.id", groupId);
+        properties.setProperty("enable.auto.commit", autoCommit);
+        properties.setProperty("auto.offset.reset",offsetReset);
+
+
+        int checkpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
+        String checkpointPath = kafkaTool.get("checkpoint.path", "file:///f/workspace/imooc-flink/state");
+
+        env.enableCheckpointing(checkpointInterval);
+        env.setStateBackend(new FsStateBackend(checkpointPath));
+        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
+
+        // 基于Kafka 原始JSON格式的字符串，获取消息流
+        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
+        DataStream<String> rawStringStream = environment.addSource(kafkaConsumer);
+        return rawStringStream;
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/window/CycleWindowAssigner.java

+package com.zhht.irn.window;
+
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.TravelEvent;
+import com.zhht.irn.trigger.CycleTimeTrigger;
+import com.zhht.irn.utils.DateUtils;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public class CycleWindowAssigner extends WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
+    @Override
+    public Collection<TimeWindow> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, WindowAssignerContext context) {
+        long start = element.getOne().getBeginDateTime().getTime();
+        long end = element.getOne().getEndDateTime().getTime();
+        return Collections.singletonList(new TimeWindow(start, end));
+    }
+
+    @Override
+    public Trigger<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
+        return new CycleTimeTrigger();
+    }
+
+    @Override
+    public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
+        return new TimeWindow.Serializer();
+    }
+
+    @Override
+    public boolean isEventTime() {
+        return true;
+    }
+}