package com.psdc.mqtt; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * mqtt配置类 */ @Configuration @Slf4j public class MqttProviderConfig { @Value("${mqtt.url}") private String url; @Value("${mqtt.username}") private String username; @Value("${mqtt.password}") private String password; @Value("${mqtt.qos}") private String qos; @Value("${mqtt.clientId}") private String clientId; @Value("${mqtt.timeOut}") private int timeOut; @Value("${mqtt.keepAlive}") private int keepAlive; @Value("${mqtt.dataTopic}") private String dataTopic; @Value("${mqtt.controlTopic}") private String controlTopic; @Value("${mqtt.controlResponseTopic}") private String controlResponseTopic; /** * 客户端对象 */ private MqttClient client; /** * 客户端连接服务端 * * @param * @return void * @date 2021/7/30 16:01 */ @Bean public MyMQTTClient connect() { //创建MQTT客户端对象 MyMQTTClient myMQTTClient = new MyMQTTClient(url, username, password, clientId, timeOut, keepAlive); for (int i = 0; i < 10; i++) { try { myMQTTClient.connect(); //不同的主题 myMQTTClient.subscribe(dataTopic, 0); myMQTTClient.subscribe(controlTopic, 0); myMQTTClient.subscribe(controlResponseTopic, 0); return myMQTTClient; } catch (MqttException e) { log.error("MQTT connect exception,connect time = " + i); try { Thread.sleep(2000); } catch (InterruptedException e1) { e1.printStackTrace(); } } } return myMQTTClient; } }