物联网实战|Spring Boot MQTT平台|解决产线emqx自动重连

.sessionExpiryInterval(5, TimeUnit.MINUTES)

.cleanStart(false)                              // 尝试恢复已有会话
.sessionExpiryInterval(TimeUnit.MINUTES.toSeconds(5))  // 断开后保留会话 5 分钟

package com.jysemel.iot.client;

import com.hivemq.client.mqtt.MqttClientSslConfig;
import com.hivemq.client.mqtt.lifecycle.MqttDisconnectSource;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
import com.hivemq.client.mqtt.mqtt5.message.connect.Mqtt5Connect;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.net.ssl.TrustManagerFactory;
import java.io.InputStream;
import java.security.KeyStore;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Slf4j
@Configuration
public class MqttConfig {

    @Bean
    public  Mqtt5AsyncClient mqtt5AsyncClient() {
        // 1. 加载CA根证书，构建SSL信任配置
        MqttClientSslConfig sslConfig = buildSslConfig();
        // 2. 构建MQTT TLS客户端
        String clientId = "server_" + UUID.randomUUID().toString().substring(0, 16);
        // 认证信息必须配置在 builder 级别，否则 automaticReconnect 重连时会丢失
        // 3. 构建连接报文
        Mqtt5AsyncClient client = Mqtt5Client.builder()
                .identifier(clientId)
                .serverHost("192.168.0.19")
                .serverPort(18085)
                .sslConfig(sslConfig)
                // 认证信息放在 builder 里，重连时会自动携带
                .simpleAuth()
                .username("emqx")
                .password("123456".getBytes())
                .applySimpleAuth()
                // 自动重连配置
                .automaticReconnect()
                .initialDelay(1, TimeUnit.SECONDS)
                .maxDelay(5, TimeUnit.MINUTES)
                .applyAutomaticReconnect()
                // 连接成功监听器（包括重连成功）
                .addConnectedListener(context -> {
                    log.info("MQTT 连接已建立 | clientId={}", clientId);
                })
                // 断开连接监听器
                .addDisconnectedListener(context -> {
                    Throwable cause = context.getCause();
                    MqttDisconnectSource source = context.getSource();
                    if (cause != null) {
                        log.error("MQTT 连接断开 | clientId={} | source={} | 原因: {} | 自动重连中...", clientId, source, cause.getMessage(), cause);
                    } else if (source == MqttDisconnectSource.SERVER) {
                        log.warn("MQTT 被服务端断开 | clientId={} | source={} | 自动重连中...", clientId, source);
                    } else if (source == MqttDisconnectSource.USER) {
                        log.info("MQTT 用户主动断开 | clientId={} | source={}", clientId, source);
                    } else {
                        log.warn("MQTT 连接断开 | clientId={} | source={} | 自动重连中...", clientId, source);
                    }
                    // 用户主动断开（如应用关闭）不推送告警，仅异常断开推送
                    if (source == MqttDisconnectSource.USER) {
                        return;
                    }
                })
                .buildAsync();
        // 4. 发起异步连接
        Mqtt5Connect connectMsg = Mqtt5Connect.builder()
                // 尝试恢复已有会话
                .cleanStart(false)
                // 会话过期时间 5 分钟：断开后服务端保留会话状态（订阅、未确认消息）300 秒
                .sessionExpiryInterval(TimeUnit.MINUTES.toSeconds(5))
                .keepAlive(60)
                .build();
        // 4. 发起异步连接
        client.connect(connectMsg)
                .whenComplete((connAck, throwable) -> {
                    if (throwable != null) {
                        log.error("MQTT 初始连接失败 | clientId={} |   | 错误: {}", clientId , throwable.getMessage(), throwable);
                        return;
                    }
                    log.info("MQTT 初始连接成功 | clientId={} | reasonCode={} | sessionPresent={}", clientId, connAck.getReasonCode(), connAck.isSessionPresent());
                });

        return client;
    }

    private MqttClientSslConfig buildSslConfig() {
        try {
            KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
            try (InputStream is = MqttConfig.class.getResourceAsStream("/cert/emqx.jks")) {
                if (is == null) {
                    throw new RuntimeException("信任库文件未找到: /cert/emqx.jks");
                }
                trustStore.load(is, "123456".toCharArray());
            }
            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
            tmf.init(trustStore);
            return MqttClientSslConfig.builder()
                    .trustManagerFactory(tmf)
                    // 关键：自定义 HostnameVerifier，总是验证通过（仅测试用！）
                    .hostnameVerifier((hostname, session) -> true)
                    .build();
        } catch (Exception e) {
            throw new RuntimeException("SSL 初始化失败", e);
        }
    }
}