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

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

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.sink.TravelMetricSink;
+import com.zhht.irn.source.TravelMockSource;
 import com.zhht.irn.utils.FlinkUtils;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -22,15 +23,17 @@ import java.util.List;
 public class TravelInfoJob {
     private static final Logger logger = LoggerFactory.getLogger(TravelInfoJob.class);
 
-    public static void main(String[] args) {
+    public static void main(String[] args) throws Exception {
         // 获取流式运行环境
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
         // 通过kafka配置文件，获取String类型的实时数据流
-        DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
+//        DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
 
+        DataStream<String> rawStringStream = env.addSource(new TravelMockSource());
         // 开始计算旅行信息相关指标
         rawStringStream.map((MapFunction<String, Travel>) s -> {
+            System.out.println(s);
             // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
             Travel travel = JSON.parseObject(s,Travel.class);
 
@@ -42,7 +45,9 @@ public class TravelInfoJob {
             boolean stopped = false; // 停车状态，判断截止到处理时的停车状态 false 行驶状态（默认值），true 停车状态
 
             for(int i = 0; i < locations.size(); i ++) {
+
                 double currentSpeed = locations.get(i).getSpeed();
+                if (i == 0) minSpeed = maxSpeed = currentSpeed;
 
                 // 获取最大最小值
                 if (currentSpeed > maxSpeed) maxSpeed = currentSpeed;
@@ -59,18 +64,22 @@ public class TravelInfoJob {
                 //
                 // 停车次数 = 行驶状态 切换至 停车状态 的次数
                 // 状态 stopped 从行驶状态切换停车状态一次, 记一次实际停车数 @宇昂提供的计算思路
-                if(stopped && currentSpeed >= 10) {
-                    // 停车状态下，只关注速度上限, 10km/h, 累计停车时长
+                if(stopped) {
+                    // 停车状态下，只关注速度上限, 10km/h
+                    // 如果当前速度小于 10km/h 算上一次停车的连续停车，只用累计停车时长，不用更改停车状态
+                    if(currentSpeed >= 10) {
                         stopped = false;
+                    }
                     stopTime += (locations.get(i).getDtTranjectory().getTime()
                            - locations.get(i - 1).getDtTranjectory().getTime());
-                } else
-                if(!stopped && currentSpeed < 3) {
+                } else {
                     // 行驶状态下，只关注速度下限, 3km/h, 增加停车次数
+                    if (currentSpeed < 3){
                         stopped = true;
                         stopCount ++;
                     }
                 }
+            }
 
             // 获取5个实时旅行信息指标
             travel.setStopCount(stopCount);
@@ -84,6 +93,7 @@ public class TravelInfoJob {
 
             return travel;
         }).addSink(new TravelMetricSink());
-    }
 
+        env.execute();
+    }
 }

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

@@ -8,6 +8,7 @@ import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
+import java.util.Date;
 
 /**
  * domain traffic
@@ -19,7 +20,7 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         // path暂定为固定值
-        String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
+        String path = "F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
         connection = MySQLUtils.getConnection(path);
         String insertSql="" +
                 "insert into app_travel_metric" +
@@ -43,18 +44,23 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     @Override
     public void invoke(Travel travel, Context context) throws Exception {
         System.out.println("=====invoke======" + travel.getId() + "-->" + travel.getRecordId());
-        insertPstmt.setString(1, travel.getRecordId());
-        insertPstmt.setString(2, travel.getCrossId());
-        insertPstmt.setLong(3, travel.getCarId());
+        insertPstmt.setLong(1, travel.getId());
+        insertPstmt.setLong(2, travel.getCarId());
+        insertPstmt.setString(3, travel.getCrossId());
         insertPstmt.setString(4,travel.getPlate());
-        insertPstmt.setString(5,travel.getFirstTime());
-        insertPstmt.setString(6,travel.getLastTime());
-        insertPstmt.setInt(7,travel.getStopCount());
-        insertPstmt.setDouble(8,travel.getStopTime());
-        insertPstmt.setDouble(9,travel.getAveSpeed());
-        insertPstmt.setDouble(10,travel.getMinSpeed());
-        insertPstmt.setDouble(11,travel.getMaxSpeed());
-        insertPstmt.setString(12, DateUtils.getTodayTime());
+        insertPstmt.setString(5,travel.getRecordId());
+        insertPstmt.setString(6, DateUtils.dateToString(new Date(Long.parseLong(travel.getFirstTime())), "yyyy-MM-dd HH:mm:ss"));
+        insertPstmt.setString(7,DateUtils.dateToString(new Date(Long.parseLong(travel.getLastTime())), "yyyy-MM-dd HH:mm:ss"));
+        insertPstmt.setString(8,DateUtils.dateToString(new Date(Long.parseLong(travel.getArrivedTime())), "yyyy-MM-dd HH:mm:ss"));
+        insertPstmt.setString(9,DateUtils.dateToString(new Date(Long.parseLong(travel.getInCrossTime())), "yyyy-MM-dd HH:mm:ss"));
+        insertPstmt.setString(10,DateUtils.dateToString(new Date(Long.parseLong(travel.getOutCrossTime())), "yyyy-MM-dd HH:mm:ss"));
+        insertPstmt.setString(11,DateUtils.dateToString(new Date(Long.parseLong(travel.getAwayTime())), "yyyy-MM-dd HH:mm:ss"));
+        insertPstmt.setInt(12,travel.getStopCount());
+        insertPstmt.setDouble(13,travel.getStopTime());
+        insertPstmt.setDouble(14,travel.getAvgSpeed());
+        insertPstmt.setDouble(15,travel.getMinSpeed());
+        insertPstmt.setDouble(16,travel.getMaxSpeed());
+        insertPstmt.setString(17, DateUtils.getTodayTime());
         insertPstmt.execute();
     }
 }

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

+package com.zhht.irn.source;
+
+import com.zhht.irn.utils.StringUtils;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+public class TravelMockSource extends RichSourceFunction<String> {
+    @Override
+    public void run(SourceContext<String> ctx) throws Exception {
+        String template = "{" +
+                "\"id\": ${id},\"crossId\": \"${crossId}\", \"recordId\": \"${recordId}\", \"recordTime\": ${recordTime}," +
+                "\"carId\": \"${carId}\", \"carColor\": \"${carColor}\", \"plate\": \"${plate}\", \"plateColor\": \"${plateColor}\"," +
+                "\"carType\": ${carType}, \"firstLineId\": \"${firstLineId}\", \"firstTime\": ${firstTime}, \"firstSpeed\": ${firstSpeed}," +
+                "\"arrivedLineId\": \"${arrivedLineId}\", \"arrivedTime\": ${arrivedTime}, \"arrivedSpeed\": ${arrivedSpeed}," +
+                "\"inCrossLineId\": \"${inCrossLineId}\", \"inCrossTime\": ${inCrossTime}, \"inSpeed\": \"${inSpeed}\"," +
+                "\"outCrossLineId\": \"${outCrossLineId}\", \"outCrossTime\": ${outCrossTime}, \"outSpeed\": \"${outSpeed}\"," +
+                "\"awayLineId\": \"${awayLineId}\", \"awayTime\": ${awayTime}, \"awaySpeed\": \"${awaySpeed}\"," +
+                "\"lastLineId\": \"${lastLineId}\", \"lastTime\": ${lastTime}, \"lastSpeed\": \"${lastSpeed}\"," +
+                "\"crossingTime\": \"${crossingTime}\", \"remark\": \"${remark}\"," +
+                "\"locations\": [" +
+                    "{\"longtitude\": ${longtitude1}, \"latitude\": ${latitude1}, \"speed\": ${speed1}, \"dtTranjectory\": ${dtTranjectory1}}," +
+                    "{\"longtitude\": ${longtitude2}, \"latitude\": ${latitude2}, \"speed\": ${speed2}, \"dtTranjectory\": ${dtTranjectory2}}," +
+                    "{\"longtitude\": ${longtitude3}, \"latitude\": ${latitude3}, \"speed\": ${speed3}, \"dtTranjectory\": ${dtTranjectory3}}" +
+//                    "{\"longtitude\": ${longtitude4}, \"latitude\": ${latitude4}, \"speed\": ${speed4}, \"dtTranjectory\": ${dtTranjectory4}}," +
+//                    "{\"longtitude\": ${longtitude5}, \"latitude\": ${latitude5}, \"speed\": ${speed5}, \"dtTranjectory\": ${dtTranjectory5}}," +
+//                    "{\"longtitude\": ${longtitude6}, \"latitude\": ${latitude6}, \"speed\": ${speed6}, \"dtTranjectory\": ${dtTranjectory6}}," +
+//                    "{\"longtitude\": ${longtitude7}, \"latitude\": ${latitude7}, \"speed\": ${speed7}, \"dtTranjectory\": ${dtTranjectory7}}," +
+//                    "{\"longtitude\": ${longtitude8}, \"latitude\": ${latitude8}, \"speed\": ${speed8}, \"dtTranjectory\": ${dtTranjectory8}}," +
+//                    "{\"longtitude\": ${longtitude9}, \"latitude\": ${latitude9}, \"speed\": ${speed9}, \"dtTranjectory\": ${dtTranjectory9}}," +
+//                    "{\"longtitude\": ${longtitude10}, \"latitude\": ${latitude10}, \"speed\": ${speed10}, \"dtTranjectory\": ${dtTranjectory10}}," +
+//                    "{\"longtitude\": ${longtitude11}, \"latitude\": ${latitude11}, \"speed\": ${speed11}, \"dtTranjectory\": ${dtTranjectory11}}," +
+//                    "{\"longtitude\": ${longtitude12}, \"latitude\": ${latitude12}, \"speed\": ${speed12}, \"dtTranjectory\": ${dtTranjectory12}}," +
+//                    "{\"longtitude\": ${longtitude13}, \"latitude\": ${latitude13}, \"speed\": ${speed13}, \"dtTranjectory\": ${dtTranjectory13}}," +
+//                    "{\"longtitude\": ${longtitude14}, \"latitude\": ${latitude14}, \"speed\": ${speed14}, \"dtTranjectory\": ${dtTranjectory14}}," +
+//                    "{\"longtitude\": ${longtitude15}, \"latitude\": ${latitude15}, \"speed\": ${speed15}, \"dtTranjectory\": ${dtTranjectory15}}," +
+//                    "{\"longtitude\": ${longtitude16}, \"latitude\": ${latitude16}, \"speed\": ${speed16}, \"dtTranjectory\": ${dtTranjectory16}}," +
+//                    "{\"longtitude\": ${longtitude17}, \"latitude\": ${latitude17}, \"speed\": ${speed17}, \"dtTranjectory\": ${dtTranjectory17}}" +
+                "]}";
+
+        // 设置参数
+        Map<String, Object> params = new HashMap<>();
+        params.put("id", 1);
+        params.put("crossId", "13070200137");
+        params.put("recordId", "00001");
+        params.put("recordTime", 1667121436000L);
+        params.put("carId", "10");
+        params.put("carColor", "");
+        params.put("plate", "");
+        params.put("plateColor", "");
+        params.put("carType", "0");
+        params.put("firstLineId", "");
+        params.put("firstTime", 1667121422000L);
+        params.put("firstSpeed", 56.8609);
+        params.put("arrivedLineId", "11");
+        params.put("arrivedTime", 1667121426000L);
+        params.put("arrivedSpeed", 58.5416);
+
+        params.put("inCrossLineId", "12");
+        params.put("inCrossTime", 1667121434000L);
+        params.put("inSpeed", 58.7933);
+
+        params.put("outCrossLineId", "");
+        params.put("outCrossTime", 0);
+        params.put("outSpeed", 0);
+
+        params.put("awayLineId", "");
+        params.put("awayTime", 0);
+        params.put("awaySpeed", 0);
+
+        params.put("lastLineId", "");
+        params.put("lastTime", 1667121436000L);
+        params.put("lastSpeed", 58.6742);
+
+        params.put("crossingTime", 0);
+        params.put("remark", "");
+
+        // 设置速度
+        params.put("longtitude1", 114.93392);
+        params.put("latitude1", 40.748466);
+        params.put("speed1", 56.836037);
+        params.put("dtTranjectory1", 1667121422177L);
+        params.put("lineId1", "");
+
+        params.put("longtitude2", 114.93391);
+        params.put("latitude2", 40.748466);
+        params.put("speed2", 56.735176);
+        params.put("dtTranjectory2", 1667121422000L);
+        params.put("lineId2", "");
+
+        params.put("longtitude3", 114.93392);
+        params.put("latitude3", 40.748466);
+        params.put("speed3", 56.836037);
+        params.put("dtTranjectory3", 1667121422000L);
+        params.put("lineId3", "");
+
+        String s = StringUtils.replaceMarco(template, params);
+        ctx.collect(s);
+    }
+
+    @Override
+    public void cancel() {
+
+    }
+}

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

 package com.zhht.irn.utils;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class StringUtils {
+    private static String placeholderKey = "\\$\\{[\\s0-9a-zA-Z_.\\-\\s]+\\}";
 
     /**
      * 判断字符串是否为空
@@ -97,9 +102,6 @@ public class StringUtils {
         return null;
     }
 
-    public static void main(String[] args) {
-        System.out.println(setFieldInConcatString("name=xuruyun,age=55",",","name","bochen"));
-    }
 
     /**
      * 从拼接的字符串中给字段设置值
@@ -177,4 +179,52 @@ public class StringUtils {
 
     }
 
+    /**
+     * 处理特殊字符
+     * @param s
+     * @return
+     */
+    public static String escapeQueryChars(String s) {
+        if (s == null || "".equals(s)) {
+            return s;
+        }
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            if (c == '\\' || c == '+' || c == '-' || c == '!' || c == '(' || c == ')'
+                    || c == ':' || c == '^' || c == '[' || c == ']' || c == '\"'
+                    || c == '{' || c == '}' || c == '~' || c == '*' || c == '?'
+                    || c == '|' || c == '&' || c == ';' || c == '/' || c == '.'
+                    || c == '$' || Character.isWhitespace(c)) {
+                sb.append('\\');
+            }
+            sb.append(c);
+        }
+        return sb.toString();
+    }
+
+
+    public static String replaceMarco(String str, Map<String, Object> params) {
+        // list 用于存储获取到的宏变量
+        List<String> list = new ArrayList<>();
+
+        // 利用正则匹配所有的宏变量
+        Pattern macroPattern = Pattern.compile(placeholderKey);
+        Matcher matcher = macroPattern.matcher(str);
+
+        // 将所有匹配到的宏变量放入list
+        while (matcher.find()) {
+            String target = matcher.group();
+            list.add(target);
+        }
+
+        // 根据变量名从参数列表中获取数据替换掉宏变量
+        for(String target : list) {
+            String key = target.substring(2, target.length() - 1);
+            if( params != null && params.get(key) != null) {
+                str = str.replaceAll(escapeQueryChars(target), params.get(escapeQueryChars(key)).toString());
+            }
+        }
+        return str;
+    }
 }