连接MQTT增加账号信息

src/main/java/com/esv/flink/EmqClient.java

@@ -6,6 +6,8 @@ import org.fusesource.mqtt.client.MQTT;
 import org.fusesource.mqtt.client.QoS;
 import org.fusesource.mqtt.client.Topic;
 
+import java.util.Properties;
+
 /**
  * @description:
  * @project: emqdemo
@@ -19,11 +21,17 @@ import org.fusesource.mqtt.client.Topic;
 public class EmqClient {
 
     private String broker;
-    private String subTopic = "$esv/iot/#";
-    private String clientId = "subscribe_emqx_flink";
+    private String subTopic;
+    private String clientId;
+    private String username;
+    private String password;
 
-    public EmqClient(String broker) {
-        this.broker = broker;
+    public EmqClient(Properties properties) {
+        this.broker = properties.getProperty("mqtt.host");
+        this.subTopic = properties.getProperty("mqtt.subscribe.topic");
+        this.clientId = properties.getProperty("mqtt.client.id");
+        this.username = properties.getProperty("mqtt.username");
+        this.password = properties.getProperty("mqtt.password");
     }
 
     public FutureConnection run () {
@@ -36,6 +44,8 @@ public class EmqClient {
             mqtt.setKeepAlive((short) 30);
             mqtt.setSendBufferSize(64);
             mqtt.setClientId(clientId);
+            mqtt.setUserName(username);
+            mqtt.setPassword(password);
 
             Topic[] topics = {
                     new Topic(subTopic, QoS.AT_MOST_ONCE)
@@ -44,13 +54,11 @@ public class EmqClient {
             FutureConnection connection = mqtt.futureConnection();
             connection.connect();
             connection.subscribe(topics);
-
-            log.info("mqtt连接成功");
+            log.info("mqtt连接成功: {}", broker);
 
             return connection;
-
-        } catch (Exception me) {
-            me.printStackTrace();
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
         }
         return null;
     }

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

@@ -77,7 +77,7 @@ public class StreamingJob {
         String brokerUrl = properties.getProperty("broker.url", "tcp://192.168.0.122:1883");
 
 		env.setParallelism(1);
-        EmqSource emqSource = new EmqSource(brokerUrl);
+        EmqSource emqSource = new EmqSource(properties);
 		DataStream<String> inStream = env.addSource(emqSource);
 		inStream.print();
 
@@ -113,20 +113,20 @@ public class StreamingJob {
 //		res.writeAsText("D://flink_result.txt");
 
 		// execute program
-		env.execute("Streaming WordCount");
+		env.execute("Streaming DataCenter IOT.");
 	}
 
 	public static class EmqSource implements ParallelSourceFunction<String> {
 		private static final long serialVersionUID = 1L;
 		private volatile boolean isRunning = true;
-		private String brokerUrl;
+		private Properties properties;
 
-		public EmqSource(String brokerUrl) {
-		    this.brokerUrl = brokerUrl;
+		public EmqSource(Properties properties) {
+			this.properties = properties;
         }
 
 		public void run(SourceContext<String> ctx) throws Exception {
-			EmqClient emqClient = new EmqClient(brokerUrl);
+			EmqClient emqClient = new EmqClient(properties);
 			FutureConnection connection = emqClient.run();
 
 			int num = 0;

src/main/resources/application-dev.properties

-broker.url= tcp://192.168.31.248:1883
+mqtt.host=tcp://192.168.31.248:1883
+mqtt.username=esv_mqtt_server
+mqtt.password=123456
+mqtt.subscribe.topic=$esv/iot/#
+mqtt.client.id=subscribe_emqx_flink
 postgres.url=jdbc:postgresql://192.168.31.248:5432/iot
 postgres.user=iot
 postgres.pwd=123456

src/main/resources/application-test.properties

-broker.url= tcp://192.168.0.122:1883
+mqtt.host=tcp://192.168.0.122:1883
+mqtt.username=esv_mqtt_server
+mqtt.password=123456
+mqtt.subscribe.topic=$esv/iot/#
+mqtt.client.id=subscribe_emqx_flink
 postgres.url=jdbc:postgresql://192.168.0.17:54321/iot
 postgres.user=iot
 postgres.pwd=123456