[代码修改]修改kakfa消费程序的方法

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

@@ -545,7 +545,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         BigDecimal bd2 = new BigDecimal(v);
         Double d3 = bd2.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
         laneNorm.setPassCapacity(d3);
-        //饱和流率
+        //饱和度
         Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime));
         BigDecimal bd1 = new BigDecimal(laneEffectGreenTime==0d?0:saturation);
         Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
@@ -562,7 +562,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         //计算空间占有长度
         laneNorm.setSpotSpaceLength(maxQueueLength);
         if("debug".equals(loggerLevel)){
-            //System.out.println("车道指标信息->"+laneNorm.toString()+" 车道饱和度->"+saturation_flow_rate+" 有效绿灯时间为->"+effcetGreenTime[1]+" 周期有效绿灯时间->"+cycle.getDuration());
+            System.out.println("车道指标信息->"+laneNorm.toString()+" 车道饱和度->"+saturation_flow_rate+" 有效绿灯时间为->"+effcetGreenTime[1]+" 周期有效绿灯时间->"+cycle.getDuration());
         }
         return laneNorm;
     }
@@ -606,7 +606,6 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         } catch (Exception e) {
 
         }
-
         if (i==0 || aLong == 0) {
             return 0d;
         }

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

@@ -35,8 +35,8 @@ public class LaneNormJob {
             //env.setRestartStrategy(RestartStrategies.noRestart());
             //设置重启策略
             //env.setRestartStrategy( RestartStrategies.fixedDelayRestart(10, Time.seconds(10)));
-            DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"signal_cycle_data","laneNormJob1");
-            DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"trips_event_info","laneNormJob1");
+            DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob1","latest");
+            DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob1","latest");
             //测试用topic
             //DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"s_data2","laneNormJob1");
             //DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"t_info2","laneNormJob1");