<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-mqtt</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-configuration-processor</artifactId>
  <optional>true</optional>
</dependency>

mqtt:
  #MQTT-服務(wù)器連接地址，如果有多個，用逗號隔開
  host: tcp://192.168.17.101:1883
  #MQTT-連接服務(wù)器默認(rèn)客戶端ID，可以隨便寫
  clientId: mqtt_test
  #MQTT-用戶名
  username: zhangsan
  #MQTT-密碼
  password: 123456
  #MQTT-指定消息的推送和訂閱主題
  topic: test
  #連接超時
  timeout: 100
  #設(shè)置會話心跳時間
  keepalive: 10

@Slf4j
@Configuration
@ConfigurationProperties("mqtt")
@Data
public class MqttConfig {
    String host;
    String clientId;
    String topic;
    String username;
    String password;
    Integer timeout;
    Integer keepalive;
    // MQTT客戶端的配置類，可以設(shè)置mqtt服務(wù)器的賬號和密碼
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        // 設(shè)置是否自動重連
        options.setAutomaticReconnect(true);
        // false 保持會話不被清理自動重連后才能收到訂閱的主題消息（包括離線時發(fā)布的消息）
        options.setCleanSession(true);
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        return options;
    }
    // MqttClient 類，MQTT的客戶端類，可以去連接MQTT服務(wù)器
    @Bean
    public MqttClient mqttClient(MqttConnectOptions mqttConnectOptions) {
        try {
            MqttClient client = new MqttClient(host, clientId);
            // 回調(diào)對象，監(jiān)聽消息的獲取，采用的接口回調(diào)，可以獲取對應(yīng)訂閱到的消息
            client.setCallback(new MessageCallback(client, this.topic, mqttConnectOptions));
            // 連接
            client.connect(mqttConnectOptions());
            return client;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("mqtt 連接異常");
        }
    }
}

/**
 * consumer 消費(fèi)者，對收到的消息進(jìn)行處理
 */
//@Component
@Slf4j
public class MessageCallback implements MqttCallbackExtended {
    private MqttClient client;
    private String topic;
    private MqttConnectOptions mqttConnectOptions;
    public MessageCallback() {
    }
    public MessageCallback(MqttClient mqttClient, String topic, MqttConnectOptions mqttConnectOptions) {
        this.client = mqttClient;
        this.topic = topic;
        this.mqttConnectOptions = mqttConnectOptions;
    }
    // 在客戶端連接斷開時觸發(fā)
    @Override
    public void connectionLost(Throwable throwable) {
        if (client != null && !client.isConnected()) {
            log.info("{}, 連接斷開，正在reconnect....", client.getClientId());
            try {
                client.reconnect();
                // client.connect(this.mqttConnectOptions);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            log.info("未知異常，連接斷開");
        }
    }
    // 在客戶端與服務(wù)器連接成功時觸發(fā)
    @Override
    public void connectComplete(boolean b, String url) {
        log.info("{} 上線了{(lán)} {}", client.getClientId(), b, url);
        try {
            client.subscribe(this.topic, 0);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    // 在客戶端收到訂閱的消息時觸發(fā)
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("接收消息主題 : " + topic);
        log.info("接收消息內(nèi)容 : " + new String(message.getPayload()));
        String msg = new String(message.getPayload());
        try {
            JSONObject jsonObject = JSON.parseObject(msg);
            String clientId = String.valueOf(jsonObject.get("clientid"));
            if (topic.endsWith("disconnected")) {
                log.info("設(shè)備{}已掉線", clientId);
            } else if (topic.endsWith("connected")) {
                log.info("設(shè)備{}已上線", clientId);
            } else {
                log.info("其他主題的消息");
            }
        } catch (JSONException e) {
            log.error("JSON Format Parsing Exception : {}", msg);
        }
    }
    // 在客戶端發(fā)送消息至服務(wù)器成功時觸發(fā)
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("deliveryComplete---------" + token.isComplete());
    }
}

@Component
@Slf4j
public class MqttUtil {
    @Autowired(required = false)
    private MqttClient client;
    /**
     * 訂閱主題
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 訂閱主題
     *
     * @param topic
     */
    public void subscribe(String topic) {
        try {
            client.subscribe(topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 發(fā)布消息
     *
     * @param qos         連接方式 0,1,2 默認(rèn)0
     * @param retained    是否保留最新的消息
     * @param topic       訂閱主題
     * @param pushMessage 消息體
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            // 發(fā)送消息
            token = mqttTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 發(fā)布消息
     *
     * @param topic       主題
     * @param pushMessage 消息內(nèi)容
     */
    public void publish(String topic, String pushMessage) {
        publish(0, true, topic, pushMessage);
    }
}

@RestController
@Slf4j
public class MqttController {
    @Autowired
    MqttClient client;
    @Autowired
    MqttUtil mqttUtil;
    @GetMapping("/send")
    public String send() {
        try {
            for (int i = 0; i < 3; i++) {
                mqttUtil.publish("test", "消息hello" + i);
                log.info("發(fā)送成功：{}", i);
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "SUCCESS";
    }
}