<!-- MQTT客戶端 -->
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

spring:
  mqtt:
    # MQTT服務(wù)端地址（tcp://ip:端口）
    broker-url: tcp://localhost:1883
    # 客戶端ID（必須唯一，建議添加隨機(jī)后綴避免沖突）
    client-id: springboot-mqtt-client-${random.uuid}
    # 用戶名（Mosquitto默認(rèn)無(wú)密碼，若配置了認(rèn)證需填寫(xiě)）
    username: 
    # 密碼
    password: 
    # 默認(rèn)訂閱的主題（可配置多個(gè)，用逗號(hào)分隔）
    default-topics: test/topic, device/status
    # QoS級(jí)別（0/1/2）
    qos: 1
    # 是否自動(dòng)重連
    automatic-reconnect: true
    # 連接超時(shí)時(shí)間（毫秒）
    connection-timeout: 3000
    # 保持連接心跳時(shí)間（秒）
    keep-alive-interval: 60

@Configuration
@ConfigurationProperties(prefix = "spring.mqtt") // 綁定application.yml中的配置
@Data
public class MqttConfig {

    private String brokerUrl;
    private String clientId;
    private String username;
    private String password;
    private String[] defaultTopics;
    private int qos;
    private boolean automaticReconnect;
    private int connectionTimeout;
    private int keepAliveInterval;

    /**
     * 初始化MQTT客戶端
     */
    @Bean
    public MqttClient mqttClient() throws MqttException {
        // 1. 創(chuàng)建連接選項(xiàng)
        MqttConnectOptions options = new MqttConnectOptions();
        // 設(shè)置用戶名密碼（若服務(wù)端無(wú)認(rèn)證，可省略）
        if (username != null && !username.isEmpty()) {
            options.setUserName(username);
        }
        if (password != null && !password.isEmpty()) {
            options.setPassword(password.toCharArray());
        }
        // 設(shè)置自動(dòng)重連、連接超時(shí)、心跳時(shí)間
        options.setAutomaticReconnect(automaticReconnect);
        options.setConnectionTimeout(connectionTimeout);
        options.setKeepAliveInterval(keepAliveInterval);
        // 禁用“清除會(huì)話”（確保斷開(kāi)重連后，未接收的消息能繼續(xù)接收）
        options.setCleanSession(false);

        // 2. 創(chuàng)建MqttClient實(shí)例（MemoryPersistence表示消息持久化到內(nèi)存）
        MqttClient client = new MqttClient(brokerUrl, clientId, new MemoryPersistence());

        // 3. 設(shè)置客戶端回調(diào)（處理連接成功、消息到達(dá)、連接丟失等事件）
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                // 連接丟失時(shí)觸發(fā)（可在這里實(shí)現(xiàn)重連邏輯，不過(guò)options已配置自動(dòng)重連）
                System.out.println("MQTT連接丟失：" + cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                // 收到訂閱的消息時(shí)觸發(fā)
                System.out.println("收到MQTT消息：");
                System.out.println("主題：" + topic);
                System.out.println("內(nèi)容：" + new String(message.getPayload()));
                System.out.println("QoS級(jí)別：" + message.getQos());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                // 消息發(fā)布完成時(shí)觸發(fā)（僅QoS>0時(shí)有效）
                try {
                    if (token.isComplete()) {
                        System.out.println("消息發(fā)布成功：" + token.getMessageId());
                    }
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        });

        // 4. 連接服務(wù)端并訂閱默認(rèn)主題
        client.connect(options);
        if (defaultTopics != null && defaultTopics.length > 0) {
            // 訂閱多個(gè)主題（第二個(gè)參數(shù)為QoS數(shù)組，與主題數(shù)組一一對(duì)應(yīng)）
            int[] qosArray = new int[defaultTopics.length];
            for (int i = 0; i < defaultTopics.length; i++) {
                qosArray[i] = qos;
            }
            client.subscribe(defaultTopics, qosArray);
            System.out.println("MQTT連接成功，已訂閱主題：" + String.join(",", defaultTopics));
        }

        return client;
    }
}

@Component
public class MqttUtil {

    @Resource
    private MqttClient mqttClient;

    /**
     * 發(fā)布消息
     * @param topic 主題
     * @param payload 消息內(nèi)容
     * @param qos QoS級(jí)別（0/1/2）
     * @param retained 是否保留消息（true：服務(wù)端保留最新消息，新訂閱者可獲取）
     */
    public void publish(String topic, String payload, int qos, boolean retained) throws MqttException {
        if (!mqttClient.isConnected()) {
            mqttClient.reconnect(); // 若連接斷開(kāi)，先重連
        }
        // 創(chuàng)建MQTT消息
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        // 發(fā)布消息
        mqttClient.publish(topic, message);
    }

    /**
     * 訂閱主題（重載方法，使用默認(rèn)QoS）
     */
    public void subscribe(String topic) throws MqttException {
        subscribe(topic, 1); // 默認(rèn)QoS=1
    }

    /**
     * 訂閱主題
     */
    public void subscribe(String topic, int qos) throws MqttException {
        if (!mqttClient.isConnected()) {
            mqttClient.reconnect();
        }
        mqttClient.subscribe(topic, qos);
        System.out.println("已訂閱主題：" + topic + "（QoS：" + qos + "）");
    }

    /**
     * 取消訂閱主題
     */
    public void unsubscribe(String topic) throws MqttException {
        if (mqttClient.isConnected()) {
            mqttClient.unsubscribe(topic);
            System.out.println("已取消訂閱主題：" + topic);
        }
    }
}

@RestController
@RequestMapping("/mqtt")
public class MqttController {

    @Resource
    private MqttUtil mqttUtil;

    /**
     * 發(fā)布MQTT消息接口
     * @param topic 主題（如test/topic）
     * @param message 消息內(nèi)容
     * @param qos QoS級(jí)別（0/1/2，默認(rèn)1）
     * @return 發(fā)布結(jié)果
     */
    @PostMapping("/publish")
    public String publish(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam(required = false, defaultValue = "1") int qos) {
        try {
            // 發(fā)布消息（retained=false：不保留消息）
            mqttUtil.publish(topic, message, qos, false);
            return"消息發(fā)布成功！主題：" + topic + "，內(nèi)容：" + message;
        } catch (MqttException e) {
            e.printStackTrace();
            return"消息發(fā)布失敗：" + e.getMessage();
        }
    }
}