删除绿信比

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

-package com.zhht.irn.flow;
-
-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\":\"2\",\"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