物联网实战：Spring Boot MQTT | 模拟器Paho客户端拆解核心点

package com.jysemel.iot;

import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;

import java.nio.charset.StandardCharsets;
import java.util.Random;

/**
 * 规范 + 高性能 + 生产可靠
 * MQTT v5 发布者（自动重连、心跳保活、无资源泄漏）
 */
public class MqttSensorPublisherReConnect {

    private static MqttClient client;
    private static MqttConnectionOptions options;

    private static final Random random = new Random();
    private static final int KEEP_ALIVE = 10;          // 心跳10秒（规范）
    private static final int PUB_INTERVAL = 2000;      // 发布间隔2秒
    private static final int RECONNECT_INTERVAL = 2000;// 重连间隔2秒（避免频繁攻击服务器）

    public static void main(String[] args) {
        startPublishLoop();
        simulateAbortDisconnect(); // 测试用：模拟异常断开
    }

    // ====================== 【规范】发布主循环（高性能、低CPU） ======================
    private static void startPublishLoop() {
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // 【规范】连接状态校验 + 自动重连
                    if (client == null || !client.isConnected()) {
                        closeClient();  // 【规范】先释放旧资源
                        connect();      // 重建连接
                    }

                    // 【规范】发布消息
                    publishSensorData();

                } catch (Exception e) {
                    System.err.println("【异常】连接异常，准备重连...");
                    client = null; // 标记重连
                }

                // 【性能】固定间隔，降低CPU占用
                sleep(PUB_INTERVAL);
            }
        }, "mqtt-publish-thread").start();
    }

    // ====================== 【规范】发布业务 ======================
    private static void publishSensorData() throws MqttException {
        String temp = "温度：" + String.format("%.1f", 20 + random.nextFloat() * 10);
        MqttMessage msg = new MqttMessage(temp.getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);          // 【规范】工业级QoS
        msg.setRetained(false); // 【规范】传感器数据不保留
        client.publish(MqttConst.TOPIC_TEMP, msg);
        System.out.println("已发布：" + temp);
    }

    // ====================== 【规范】连接方法 ======================
    private static void connect() {
        try {
            client = new MqttClient(MqttConst.BROKER, MqttConst.PUBLISH_CLIENT_ID, new MemoryPersistence());
            options = new MqttConnectionOptions();

            // MQTT v5 标准连接配置
            options.setKeepAliveInterval(KEEP_ALIVE);
            options.setConnectionTimeout(5);
            options.setCleanStart(true);
            options.setUserName(MqttConst.USERNAME);
            options.setPassword(MqttConst.PASSWORD.getBytes(StandardCharsets.UTF_8));

            client.connect(options);
            System.out.println("【发布者】连接成功 ✅");

        } catch (Exception e) {
            System.err.println("连接失败，" + (RECONNECT_INTERVAL / 1000) + "秒后重试...");
            sleep(RECONNECT_INTERVAL);
        }
    }

    // ====================== 【规范】关闭客户端（防止资源泄漏） ======================
    private static void closeClient() {
        if (client != null) {
            try {
                if (client.isConnected()) {
                    client.disconnect();
                }
                client.close();
            } catch (MqttException ignored) {}
        }
    }

    // ====================== 【测试】模拟异常断开 ======================
    private static void simulateAbortDisconnect() {
        new Thread(() -> {
            sleep(10000);
            System.err.println("\n===== 模拟异常断开 =====");
            try {
                if (client != null && client.isConnected()) {
                    client.disconnectForcibly(0);
                }
            } catch (Exception ignored) {}
        }, "mqtt-simulate-disconnect").start();
    }

    // ====================== 【工具】睡眠方法 ======================
    private static void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException ignored) {
            Thread.currentThread().interrupt();
        }
    }
}