完善实时计算指标旅行信息相关的5个指标的计算逻辑代码

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

@@ -106,12 +106,16 @@ public class Travel {
 
     //实际停车次数
     private int stopCount;
+
     //停车总时长
     private double stopTime;
+
     //平均车速
-    private double aveSpeed;
+    private double avgSpeed;
+
     //最低速度
     private double minSpeed;
+
     //最高速度
     private double maxSpeed;
 
@@ -371,12 +375,12 @@ public class Travel {
         this.stopTime = stopTime;
     }
 
-    public double getAveSpeed() {
-        return aveSpeed;
+    public double getAvgSpeed() {
+        return avgSpeed;
     }
 
-    public void setAveSpeed(double aveSpeed) {
-        this.aveSpeed = aveSpeed;
+    public void setAvgSpeed(double avgSpeed) {
+        this.avgSpeed = avgSpeed;
     }
 
     public double getMinSpeed() {

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

@@ -3,24 +3,15 @@ package com.zhht.irn.job;
 import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
-import com.zhht.irn.entity.mysql.TripsMetric;
 import com.zhht.irn.sink.TravelMetricSink;
-import org.apache.flink.api.common.functions.FilterFunction;
+import com.zhht.irn.utils.FlinkUtils;
 import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.common.serialization.SimpleStringSchema;
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.runtime.state.filesystem.FsStateBackend;
 import org.apache.flink.streaming.api.datastream.DataStream;
-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.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
+
+import java.util.List;
 
 /**
  * 旅行信心实时任务
@@ -34,92 +25,65 @@ public class TravelInfoJob {
     public static void main(String[] args) {
         // 获取流式运行环境
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        // 获取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 = env.addSource(kafkaConsumer);
+
+        // 通过kafka配置文件，获取String类型的实时数据流
+        DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
+
         // 开始计算旅行信息相关指标
-        rawStringStream.map(new MapFunction<String, Travel>() {
-            @Override
-            public Travel map(String s) throws Exception {
-                // json 格式字符串转化为 TravelInfo 实体类
-                Travel travel = JSON.parseObject(s,Travel.class);
-                String crossId = travel.getCrossId();
-                String recordId = travel.getRecordId();
-                long carId = travel.getCarId();
-                List<Location> locations = travel.getLocations();
-                //两个location之间的间隔时长0.1s
-                double intervalTime = 0.1;
-                //停车次数
-                int stopCount = 0;
-                //停车总时长
-                double stopTime= 0;
-                //速度总和
-                double totalSpeed=0;
-                //速度数组
-                List<Double> speedList = new ArrayList<Double>();
-                //上一个location的速度,初始化取第一条数据
-                double lastSpeed=locations.get(0).getSpeed();
-                for(int i=0;i<locations.size();i++){
-                    Location location = locations.get(i);
-                    Double speed = location.getSpeed();
-                    speedList.add(speed);
-                    if(speed <= 3){
-                        if(lastSpeed>3) {
-                            stopCount = stopCount + 1;
-                        }
-                        stopTime = stopTime + intervalTime;
-                    }
-                    totalSpeed = totalSpeed + speed;
-                    lastSpeed = speed;
+        rawStringStream.map((MapFunction<String, Travel>) s -> {
+            // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
+            Travel travel = JSON.parseObject(s,Travel.class);
+
+            List<Location> locations = travel.getLocations();
+
+            int stopCount = 0; long stopTime = 0L;
+            double minSpeed = 0.0, maxSpeed = 0.0, totalSpeed = 0.0;
+
+            boolean stopped = false; // 停车状态，判断截止到处理时的停车状态 false 行驶状态（默认值），true 停车状态
+
+            for(int i = 0; i < locations.size(); i ++) {
+                double currentSpeed = locations.get(i).getSpeed();
+
+                // 获取最大最小值
+                if (currentSpeed > maxSpeed) maxSpeed = currentSpeed;
+                if (currentSpeed < minSpeed) minSpeed = currentSpeed;
+
+                // 累计速度
+                totalSpeed += currentSpeed;
+
+                // 实际停车为整个旅行中车辆速度低于某个下限阈值的(暂定3km/h)次数. 当车辆一直低于该下限阈值(10km/h)时, 视为一次连续的实际停车.
+                // 当车辆高于某个上限阈值(暂定10km/h)后, 再次低于下限阈值(暂定3km/h)时, 再记为一次实际停车
+
+                // 每低于3 => 视为停车状态
+                // 每高于10 => 视为行驶状态
+                //
+                // 停车次数 = 行驶状态 切换至 停车状态 的次数
+                // 状态 stopped 从行驶状态切换停车状态一次, 记一次实际停车数 @宇昂提供的计算思路
+                if(stopped && currentSpeed >= 10) {
+                    // 停车状态下，只关注速度上限, 10km/h, 累计停车时长
+                   stopped = false;
+                   stopTime += (locations.get(i).getDtTranjectory().getTime()
+                           - locations.get(i - 1).getDtTranjectory().getTime());
+                } else
+                if(!stopped && currentSpeed < 3) {
+                    // 行驶状态下，只关注速度下限, 3km/h, 增加停车次数
+                    stopped = true;
+                    stopCount ++;
                 }
-                //最大速度
-                double  maxSpeed = Collections.max(speedList);
-                //最小速度，初始化0
-                double minSpeed = Collections.min(speedList);
-                //平均速度
-                double aveSpeed = totalSpeed/locations.size();
-                travel.setStopCount(stopCount);
-                travel.setStopTime(stopTime);
-                travel.setAveSpeed(aveSpeed);
-                travel.setMaxSpeed(maxSpeed);
-                travel.setMinSpeed(minSpeed);
-                return travel;
             }
-        }).addSink(new TravelMetricSink()); //@todo 插入数据库的逻辑
 
+            // 获取5个实时旅行信息指标
+            travel.setStopCount(stopCount);
+            travel.setStopTime(stopTime);
+            travel.setAvgSpeed(totalSpeed / locations.size());
+            travel.setMaxSpeed(maxSpeed);
+            travel.setMinSpeed(minSpeed);
+
+            logger.info("完成旅行信息的指标计算，当前旅行车辆旅行ID{}的实时指标为， 最高速度：{}， 最低速度：{}， 平均速度：{}， 停车次数： {}，" +
+                    "停车总时长：{}", travel.getRecordId(), maxSpeed, minSpeed, travel.getAvgSpeed(), stopCount, stopTime);
+
+            return travel;
+        }).addSink(new TravelMetricSink());
     }
 
 }

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

@@ -21,7 +21,11 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
         // path暂定为固定值
         String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
         connection = MySQLUtils.getConnection(path);
-        String insertSql="insert into app_travel_metric(record_id,cross_id,car_id,plate,capture_time,lost_time,stop_count,stop_time,ave_speed,min_speed,max_speed,update_time) values (?,?,?,?,?,?,?,?,?,?,?,?)";
+        String insertSql="" +
+                "insert into app_travel_metric" +
+                "(id, car_id, cross_id, plate, record_id, capture_time, lost_time, arrived_time, incross_time, " +
+                "outcross_time, away_time, stop_count, stop_time, avg_speed, min_speed, max_speed, update_time)" +
+                "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         insertPstmt = connection.prepareStatement(insertSql);
     }
     @Override