在物联网（IoT）和智能设备横行的今天，你有没有遇到这样的问题：

• 服务端需要实时把报警、状态更新、控制指令推送给客户端；
• 安卓 App、嵌入式设备、网页等终端，需要轻量且稳定的连接方式；
• HTTP 太“重”、WebSocket 配置又麻烦？

这时，轻量级消息传输协议 MQTT（Message Queuing Telemetry Transport）登场！

一句话理解 MQTT：专为低带宽、高并发、实时通信设计的发布-订阅协议。

那么问题来了 —— 在 Spring Boot 项目中，如何快速、优雅、高可控地落地 MQTT？

-01-

MQTT 接入方案选择

MQTT 本身只是一种通信协议，并不指定你用哪个消息中间件。而目前支持 MQTT 的主流 Broker 包括：

本次我们采用的是：RabbitMQ + MQTT 插件，实现服务端到安卓客户端的推送通知，配合 Spring Boot 框架，集成简便，生产可用！

-02-

MQTT 三大角色

MQTT，就像微信一样：

• Publisher（发布者）：你发朋友圈
• Broker（中间人）：微信服务器
• Subscriber（订阅者）：看到你朋友圈的朋友

也就是说，消息不是点对点的，而是“你说一句，谁订阅了就能听到”。

-03-

实战解析

Spring Boot + RabbitMQ MQTT 实现推送系统

整体架构：

[Spring Boot服务] --发布消息--> [RabbitMQ MQTT插件] --> [MQTT客户端订阅接收消息]

RabbitMQ 开启 MQTT 插件

rabbitmq-plugins enable rabbitmq_mqtt        # 服务端 MQTT 协议，端口1883
rabbitmq-plugins enable rabbitmq_web_mqtt    # Web前端用 MQTT 协议，端口15675

引入依赖

<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-mqtt</artifactId>
</dependency>

配置 application.yml

mqtt-push:
  clientId: mqtt_client_
  serverClientId: mqtt_server_
  servers: tcp://127.0.0.1:1883
  username: guest
  password: guest
  defaultTopic: sensor/+/temperature

配置连接工厂

@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setServerURIs(servers.split(","));
    options.setCleanSession(false);
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    options.setKeepAliveInterval(20);
    factory.setConnectionOptions(options);
    return factory;
}

服务端推送消息

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
    MqttPahoMessageHandler handler = new MqttPahoMessageHandler(serverClientId + "producer_" + RandomUtil.getRandomStr(), mqttClientFactory());
    handler.setAsync(true);
    handler.setDefaultQos(1);
    handler.setDefaultTopic(defaultTopic);
    return handler;
}

使用接口发送消息：

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
    void sendMessage2Mqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
}

服务端监听客户端消息

@Bean
public MessageProducer inbound() {
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
        clientId + "consumer_" + RandomUtil.getRandomStr(), mqttClientFactory(), defaultTopic);
    adapter.setQos(2);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
}

处理消息回调：

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler mqttInMessageHandler() {
    return message -> {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
        String payload = message.getPayload().toString();
        log.info("收到消息：主题 [{}] 内容 [{}]", topic, payload);
    };
}

MQTTBox 测试

MQTTBox 是一款强大的 MQTT 测试工具，可以模拟发送消息，也能订阅查看接收到的消息：

1. 发布测试

使用 MQTTBox 向 sensor/s123/temperature 发布消息
服务端通过通配符 sensor/+/temperature 成功收到消息！

2. 控制器测试

@PostMapping("/sendMessage")
public String sendMqtt(@RequestBody ReqSendMsgDTO dto) {
    mqttGateway.sendMessage2Mqtt(dto.getTopic(), dto.getPayload());
    return "SUCCESS";
}

-04-

总结

实践建议

• clientId 必须唯一，推荐使用 UUID 或服务实例标识；
• QoS 建议使用 1（至少一次），避免消息丢失；
• 若用 RabbitMQ，也可以使用 Exchange + Topic Binding 方式做高级路由；
• 对于高并发或长连接推送，推荐结合 Netty 或 Gateway 层限流处理。

技术方案