SpringBoot集成MQTT单Topic与多Topic订阅与消费

1. 适用场景

SpringBoot集成MQTT单Topic与多Topic订阅与消费功能在多个领域都有广泛应用：

物联网设备管理

• 智能家居设备状态监控与控制
• 工业传感器数据采集与处理
• 车联网实时数据传输

实时消息推送

• 移动应用推送通知
• 即时通讯消息分发
• 系统状态实时更新

分布式系统通信

• 微服务间异步通信
• 跨平台数据同步
• 事件驱动架构实现

监控与告警系统

• 服务器性能监控
• 业务指标实时统计
• 异常事件告警通知

2. 适配系统与环境配置要求

系统要求

• Java 8及以上版本
• Spring Boot 2.x或3.x版本
• Maven或Gradle构建工具

MQTT Broker要求

• 支持MQTT 3.1.1或5.0协议
• 常见Broker：EMQX、Mosquitto、HiveMQ等
• 支持TCP和SSL连接

依赖配置

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

环境变量配置

• MQTT Broker地址和端口
• 客户端ID配置
• 用户名和密码认证
• QoS级别设置
• 连接超时和保持连接间隔

3. 资源使用教程

单Topic订阅配置

配置文件设置

mqtt:
  broker-url: tcp://localhost:1883
  client-id: springboot-client
  username: admin
  password: password
  default-topic: device/status

单Topic消费者

@Configuration
public class SingleTopicConfig {
    
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{"tcp://localhost:1883"});
        options.setUserName("admin");
        options.setPassword("password".toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }
    
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter("springboot-client", 
                mqttClientFactory(), "device/status");
        adapter.setCompletionTimeout(5000);
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
    
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) {
        System.out.println("收到消息: " + message.getPayload());
    }
}

多Topic订阅配置

多Topic消费者

@Configuration
public class MultipleTopicsConfig {
    
    @Bean
    public MessageChannel mqttMultiInputChannel() {
        return new DirectChannel();
    }
    
    @Bean
    public MessageProducer multiTopicInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter("multi-client", 
                mqttClientFactory());
        
        // 订阅多个主题
        adapter.addTopic("sensor/temperature", 1);
        adapter.addTopic("sensor/humidity", 1);
        adapter.addTopic("device/+/status", 1); // 使用通配符
        
        adapter.setCompletionTimeout(5000);
        adapter.setOutputChannel(mqttMultiInputChannel());
        return adapter;
    }
    
    @ServiceActivator(inputChannel = "mqttMultiInputChannel")
    public void handleMultiTopicMessage(Message<?> message) {
        String topic = (String) message.getHeaders().get("mqtt_receivedTopic");
        Object payload = message.getPayload();
        
        System.out.println("主题: " + topic + ", 消息: " + payload);
        
        // 根据不同的主题进行不同的处理
        if (topic.startsWith("sensor/temperature")) {
            processTemperature(payload);
        } else if (topic.startsWith("sensor/humidity")) {
            processHumidity(payload);
        } else if (topic.contains("status")) {
            processDeviceStatus(payload);
        }
    }
    
    private void processTemperature(Object data) {
        // 温度数据处理逻辑
    }
    
    private void processHumidity(Object data) {
        // 湿度数据处理逻辑
    }
    
    private void processDeviceStatus(Object data) {
        // 设备状态处理逻辑
    }
}

消息发布配置

消息发布器

@Component
public class MqttMessagePublisher {
    
    @Autowired
    private MqttPahoClientFactory mqttClientFactory;
    
    private MqttTemplate mqttTemplate;
    
    @PostConstruct
    public void init() {
        mqttTemplate = new MqttTemplate(mqttClientFactory);
    }
    
    public void publish(String topic, String message) {
        mqttTemplate.send(topic, message);
    }
    
    public void publish(String topic, String message, int qos) {
        mqttTemplate.send(topic, message, qos, false);
    }
}

4. 常见问题及解决办法

连接问题

连接超时或拒绝

• 检查Broker地址和端口是否正确
• 确认网络访问设置
• 验证用户名和密码认证信息

解决方案

// 设置连接选项
MqttConnectOptions options = new MqttConnectOptions();
options.setConnectionTimeout(30); // 设置连接超时时间
options.setKeepAliveInterval(60); // 设置心跳间隔
options.setAutomaticReconnect(true); // 启用自动重连

消息丢失问题

QoS级别设置

• QoS 0：最多一次，可能丢失消息
• QoS 1：至少一次，可能重复消息
• QoS 2：恰好一次，保证消息不丢失不重复

推荐配置

// 对于重要消息使用QoS 1或2
adapter.setQos(1); // 订阅时设置QoS
mqttTemplate.send(topic, message, 1, false); // 发布时设置QoS

性能优化

连接池配置

@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setMaxInflight(1000); // 设置最大飞行消息数
    factory.setConnectionOptions(options);
    return factory;
}

线程池配置

@Bean
public TaskExecutor mqttTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(50);
    executor.setQueueCapacity(1000);
    executor.setThreadNamePrefix("mqtt-");
    executor.initialize();
    return executor;
}

异常处理

连接异常处理

@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setAutomaticReconnect(true);
    options.setMaxReconnectDelay(30000); // 最大重连延迟
    factory.setConnectionOptions(options);
    return factory;
}

消息处理异常

@ServiceActivator(inputChannel = "mqttInputChannel", 
                 adviceChain = "retryAdvice")
public void handleMessage(Message<?> message) {
    try {
        // 消息处理逻辑
    } catch (Exception e) {
        log.error("消息处理异常: {}", e.getMessage());
        // 可以选择重试或者记录到死信队列
    }
}

通过合理的配置和异常处理，SpringBoot集成MQTT可以实现稳定可靠的单Topic和多Topic消息订阅与消费，满足各种物联网和实时消息处理场景的需求。