绿信比

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

+package com.zhht.irn.job;
+
+import com.alibaba.fastjson.JSON;
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.Stage;
+import com.zhht.irn.sink.GreenRadioSink;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.List;
+
+/**
+ * 两阶段提交参考文档：
+ * https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
+ */
+public class GreenRadioJob {
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+//        DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
+        // 目前假设从Kafka接入的数据为json格式的字符串
+        // Cycle Topic 的消息格式如下：
+        // {"id":"1","crossCode":"0001","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"300","cycleOrder":"1",
+        // "stageList":[
+        //     {"phaseValue":"101","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:01:00","duration":"60","green":"30","yellow":"3","allRed":"0","phaseOrder":"1"},
+        //     {"phaseValue":"102","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"240","green":"120","yellow":"3","allRed":"0","phaseOrder":"2"}
+        //  ]
+        //}
+        // 造一条数据
+        DataStream<String> stream = env.fromElements("{\"id\":\"3\",\"crossCode\":\"0001\",\"signalCode\":\"1000001\",\"beginDateTime\":\"2020-08-02 08:10:03\",\"endDateTime\":\"2020-08-02 08:12:11\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"stageList\":[{\"phaseValue\":\"101\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endDateTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\",\"phaseOrder\":\"1\"},{\"phaseValue\":\"102\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endDateTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\",\"phaseOrder\":\"2\"}]}");
+        DataStream<Cycle> lxbStream = stream.map(new MapFunction<String, Cycle>() {
+            @Override
+            public Cycle map(String s) throws Exception {
+                //启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
+                //清场损失时间 = 黄灯时长 + 全红时长 - 2
+                //阶段损失时间 = 启动损失时间 + 清场损失时间
+                //阶段有效绿灯时间 = 绿灯时长+黄灯时长+全红时长 - 阶段损失时间
+                //阶段绿信比 = 阶段有效绿灯时间/周期时长
+                //周期绿信比 = 阶段绿信比之和
+                Cycle cycle = JSON.parseObject(s, Cycle.class);
+                long cycleId= cycle.getId();
+                int duration = cycle.getDuration();
+                String crossCode = cycle.getCrossCode();
+                double cycleGreenRatio = 0;
+                int cycleLossTime = 0;
+                List<Stage> stageList = cycle.getStageList();
+                for (Stage stage : stageList) {
+                    //相位号
+                    String phaseValue = stage.getPhaseValue();
+                    // 绿灯时间
+                    int green = stage.getGreen();
+                    // 黄灯时间
+                    int yellow = stage.getYellow();
+                    // 全红时间
+                    int allRed = stage.getAllRed();
+                    // 启动损失时间，一般定义：2s
+                    int startLossTime = 2;
+                    //清场损失时间
+                    int clearLossTime = yellow + allRed - 2;
+                    //阶段损失时间
+                    int phaseLossTime = startLossTime + clearLossTime;
+                    //阶段有效绿灯时间
+                    int phaseValidGreen = green + yellow + allRed - phaseLossTime;
+                    System.out.println("******duration:"+duration);
+                    System.out.println("******phaseValidGreen:"+phaseValidGreen);
+
+                    //阶段绿信比
+                    double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
+                    System.out.println("******phaseGreenRatio:"+phaseGreenRatio);
+                    //组装阶段绿信比、有效绿灯时间、损失时间
+                    stage.setGreenRatio(phaseGreenRatio);
+                    stage.setValidGreen(phaseValidGreen);
+                    stage.setLossTime(phaseLossTime);
+                    //周期绿信比= 各个阶段绿信比之和
+                    cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
+                    //周期损失时间= 各个阶段损失时间之和
+                    cycleLossTime = cycleLossTime + phaseLossTime;
+                    System.out.println("******cycleGreenRatio:"+cycleGreenRatio);
+                }
+                // 组装周期绿信比、损失时间
+                cycle.setGreenRatio(cycleGreenRatio);
+                cycle.setLossTime(cycleLossTime);
+                return cycle;
+            }
+        });
+        lxbStream.addSink(new GreenRadioSink());
+        env.execute();
+    }
+}
\ No newline at end of file