增加fast

realtime/hologram-streaming/src/main/java/com/zhht/irn/process/LaneNormProcessFunction.java

@@ -65,7 +65,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         Cycle f1 = cycle;
         // 代表周期已到，需要进行相应的指标计算了
         //周期内所有过车数据
-        List<TravelEvent> travelEvents = cycleTravel.get(f1.getCrossId());
+        List<TravelEvent> travelEvents = cycleTravel.get(f1.getCrossCode());
         if(travelEvents==null){
             travelEvents=new ArrayList<>();
         }
@@ -74,18 +74,18 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         Map<String, List<TravelEvent>> carForLane = getCarForLane(car);
         //车道道路流向关系
         List<Map> laneList = getDictResult(connection,"SELECT lane_id,name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
-                "                where out_in_name='in' and cross_id='"+f1.getCrossId()+"'");
+                "                where out_in_name='in' and cross_id='"+f1.getCrossCode()+"'");
         //获取所有车道信息
         List<Map> dictResult = getDictResult(connection,"select a.*,b.saturation_flow_rate from (\n" +
                 "SELECT lane_id,position,lane_length,GROUP_CONCAT(position,name) as positionName FROM dim_cnt_cross_lane_position\n" +
-                "                    where cross_id='" + f1.getCrossId() + "' and  out_in_type=1 group By lane_id,cross_id,position,lane_length)a left join app_atomic_lane_saturation_flow_rate b\n" +
-                "                    on a.lane_id=b.lane_id and b.cross_id='" + f1.getCrossId() + "'");
+                "                    where cross_id='" + f1.getCrossCode() + "' and  out_in_type=1 group By lane_id,cross_id,position,lane_length)a left join app_atomic_lane_saturation_flow_rate b\n" +
+                "                    on a.lane_id=b.lane_id and b.cross_id='" + f1.getCrossCode() + "'");
         //查询路口所有的方向
         List<Map> positionList = getDictResult(connection, "SELECT position FROM dim_cnt_cross_lane_position\n" +
-                "                where out_in_name='in' and cross_id='" + f1.getCrossId() + "' GROUP BY position");
+                "                where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position");
         //查询路口所有的流向
         List<Map> positionNameList = getDictResult(connection, " SELECT position,name FROM dim_cnt_cross_lane_position\n" +
-                "        where out_in_name='in' and cross_id='" + f1.getCrossId() + "' GROUP BY position,name");
+                "        where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position,name");
         List<Stage> stageList = f1.getStageList();
         //获取相位关系
         for (Stage stage : stageList) {
@@ -106,7 +106,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         //获取所有流向的指标
         getPositionNameNorm(list,positionNameList,cycle,out);
         //过滤过期数据 没有进入事件的数据保留到下次计算
-        cycleTravel.put(f1.getCrossId(),car.stream().filter(recode->StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList()));
+        cycleTravel.put(f1.getCrossCode(),car.stream().filter(recode->StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList()));
         //关闭mysql连接
         if(connection!=null){
             connection.close();
@@ -175,7 +175,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                 laneNorm.setOrder(0);
                 laneNorm.setCycleOrder(cycle.getCycleOrder());
                 laneNorm.setDirection(position);
-                laneNorm.setCorssId(cycle.getCrossId());
+                laneNorm.setCorssId(cycle.getCrossCode());
                 laneNorm.setType("1");
                 //获取所有这个方向得车道 交通流量为0得不参与计算
                 List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())&& recode.getLaneName().indexOf(position+name)!=-1 &&recode.getTrafficCapacity()!=0d).collect(Collectors.toList());
@@ -224,7 +224,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                 laneNorm.setOrder(1);
                 laneNorm.setCycleOrder(cycle.getCycleOrder());
                 laneNorm.setDirection(position);
-                laneNorm.setCorssId(cycle.getCrossId());
+                laneNorm.setCorssId(cycle.getCrossCode());
                 laneNorm.setType("0");
                 //获取所有这个方向得车道 交通流量为0得不参与计算
                 List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())&&recode.getTrafficCapacity()!=0d).collect(Collectors.toList());
@@ -272,7 +272,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         Integer lane_length = Integer.parseInt(String.valueOf(map.get("lane_length")));
         //车道基本信息
         laneNorm.setOrder(0);
-        laneNorm.setCorssId(cycle.getCrossId());
+        laneNorm.setCorssId(cycle.getCrossCode());
         laneNorm.setDirection(String.valueOf(map.get("position")));
         laneNorm.setLaneId(lane_id);
         laneNorm.setCycleOrder(cycle.getCycleOrder());

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

@@ -36,7 +36,7 @@ public class CirlceSource implements SourceFunction<Cycle> {
     public static Cycle getCycle(Calendar beforeTime,Date date){
         Cycle cycle = new Cycle();
         cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
-        cycle.setCrossId("13070202");
+        cycle.setCrossCode("13070202");
         cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
         cycle.setEndDateTime(DateUtils.formatTime(date));
         cycle.setDuration(120);