1、添加mqtt报错信息

2、修改监听mqtt频道前缀

src/main/java/com/censoft/flink/StreamingJob.java

@@ -45,7 +45,7 @@ public class StreamingJob {
 
         //2、接收消息并将数据转化为流
         //todo 根据场景id 拼接对应的mqtt频道名称
-        DataStream<AlgorithmPushDto> mqttStream = env.addSource(new MqttConsumer("/censoft/test/" + sceneId));
+        DataStream<AlgorithmPushDto> mqttStream = env.addSource(new MqttConsumer("/censoft/test1/" + sceneId));
 
         //3、进行处理
         //3、0 更新场景在线时间

src/main/java/com/censoft/flink/mqtt/MqttConsumer.java

@@ -58,6 +58,7 @@ public class MqttConsumer extends RichParallelSourceFunction<AlgorithmPushDto> {
             System.out.println("MQTT连接成功:" + mqttConfigBean.getClientId() + ":" + client);
         } catch (Exception e) {
             System.out.println("MQTT连接异常：" + e);
+            e.printStackTrace();
         }
     }