SpringBoot如何实现MQTT消息发送和接收
Springintegration交互逻辑对于发布者：1.消息通过消息网关发送出去，由MessageChannel的实例DirectChannel处理发送的细节。2.DirectChannel收到消息后，内部通过MessageHandler的实例MqttPahoMessageHandler发送到指定的Topic。对于订阅者：1.通过注入MessageProducerSupport的实例MqttPahoMessageDrivenChannelAdapter，实现订阅Topic和绑定消息消费的MessageC
Spring integration交互逻辑
对于发布者：
1.消息通过消息网关发送出去，由 MessageChannel 的实例 DirectChannel 处理发送的细节。
2.DirectChannel 收到消息后，内部通过 MessageHandler 的实例 MqttPahoMessageHandler 发送到指定的 Topic。
对于订阅者：
1.通过注入 MessageProducerSupport 的实例 MqttPahoMessageDrivenChannelAdapter，实现订阅 Topic 和绑定消息消费的 MessageChannel。
2.同样由 MessageChannel 的实例 DirectChannel 处理消费细节。
Channel 消息后会发送给我们自定义的 MqttInboundMessageHandler 实例进行消费。
可以看到整个处理的流程和前面将的基本一致。Spring Integration 就是抽象出了这么一套消息通信的机制，具体的通信细节由它集成的中间件来决定。
1、maven依赖
org.springframework.boot spring-boot-starter-integration 2.5.1 org.springframework.integration spring-integration-stream 5.5.5 org.springframework.integration spring-integration-mqtt 5.5.5
2、yaml配置文件
#mqtt配置 mqtt: username: 123 password: 123 #MQTT-服务器连接地址，如果有多个，用逗号隔开 url: tcp://127.0.0.1:1883 #MQTT-连接服务器默认客户端ID client: id: ${random.value} default: #MQTT-默认的消息推送主题，实际可在调用接口时指定 topic: topic,mqtt/test/# #连接超时 completionTimeout: 3000
3、mqtt生产者消费者配置类
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import java.util.Arrays;
import java.util.List;
/**
* mqtt 推送and接收 消息类
**/
@Configuration
@IntegrationComponentScan
@Slf4j
public class MqttSenderAndReceiveConfig {
private static final byte[] WILL_DATA;
static {
WILL_DATA = "offline".getBytes();
}
@Autowired
private MqttReceiveHandle mqttReceiveHandle;
@Value("${mqtt.username}")
private String username;
@Value("${mqtt.password}")
private String password;
@Value("${mqtt.url}")
private String hostUrl;
@Value("${mqtt.client.id}")
private String clientId;
@Value("${mqtt.default.topic}")
private String defaultTopic;
@Value("${mqtt.completionTimeout}")
private int completionTimeout; //连接超时
/**
* MQTT连接器选项
**/
@Bean(value = "getMqttConnectOptions")
public MqttConnectOptions getMqttConnectOptions1() {
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
mqttConnectOptions.setCleanSession(true);
// 设置超时时间 单位为秒
mqttConnectOptions.setConnectionTimeout(10);
mqttConnectOptions.setAutomaticReconnect(true);
mqttConnectOptions.setUserName(username);
mqttConnectOptions.setPassword(password.toCharArray());
mqttConnectOptions.setServerURIs(new String[]{hostUrl});
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
mqttConnectOptions.setKeepAliveInterval(10);
// 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
//mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
return mqttConnectOptions;
}
/**
* MQTT工厂
**/
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(getMqttConnectOptions1());
return factory;
}
/**
* MQTT信息通道（生产者）
**/
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
/**
* MQTT消息处理器（生产者）
**/
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId + "_producer", mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(defaultTopic);
messageHandler.setAsyncEvents(true); // 消息发送和传输完成会有异步的通知回调
//设置转换器 发送bytes数据
DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
converter.setPayloadAsBytes(true);
return messageHandler;
}
/**
* 配置client,监听的topic
* MQTT消息订阅绑定（消费者）
**/
@Bean
public MessageProducer inbound() {
List topicList = Arrays.asList(defaultTopic.trim().split(","));
String[] topics = new String[topicList.size()];
topicList.toArray(topics);
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(clientId + "_consumer", mqttClientFactory(), topics);
adapter.setCompletionTimeout(completionTimeout);
DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
converter.setPayloadAsBytes(true);
adapter.setConverter(converter);
adapter.setQos(2);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
/**
* MQTT信息通道（消费者）
**/
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
/**
* MQTT消息处理器（消费者）
**/
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message message) throws MessagingException {
//处理接收消息
mqttReceiveHandle.handle(message);
}
};
}
} 4、消息处理类
/**
* mqtt客户端消息处理类
**/
@Slf4j
@Component
public class MqttReceiveHandle {
public void handle(Message message) {
log.info("收到订阅消息: {}", message);
String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
log.info("消息主题：{}", topic);
Object payLoad = message.getPayload();
byte[] data = (byte[]) payLoad;
Packet packet = Packet.parse(data);
log.info("发送的Packet数据{}", JSON.toJSONString(packet));
}
}5、mqtt发送接口
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
/**
* mqtt发送消息
* （defaultRequestChannel = "mqttOutboundChannel" 对应config配置）
* **/
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
/**
* 发送信息到MQTT服务器
*
* @param
*/
void sendToMqttObject(@Header(MqttHeaders.TOPIC) String topic, byte[] payload);
/**
* 发送信息到MQTT服务器
*
* @param topic 主题
* @param payload 消息主体
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
/**
* 发送信息到MQTT服务器
*
* @param topic 主题
* @param qos 对消息处理的几种机制。
* 0 表示的是订阅者没收到消息不会再次发送，消息会丢失。
* 1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。
* 2 多了一次去重的动作，确保订阅者收到的消息有一次。
* @param payload 消息主体
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
/**
* 发送信息到MQTT服务器
*
* @param topic 主题
* @param payload 消息主体
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, Object payload);
/**
* 发送信息到MQTT服务器
*
* @param topic 主题
* @param payload 消息主体
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, byte[] payload);
}6、mqtt事件监听类
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class MqttListener {
/**
* 连接失败的事件通知
* @param mqttConnectionFailedEvent
*/
@EventListener(classes = MqttConnectionFailedEvent.class)
public void listenerAction(MqttConnectionFailedEvent mqttConnectionFailedEvent) {
log.info("连接失败的事件通知");
}
/**
* 已发送的事件通知
* @param mqttMessageSentEvent
*/
@EventListener(classes = MqttMessageSentEvent.class)
public void listenerAction(MqttMessageSentEvent mqttMessageSentEvent) {
log.info("已发送的事件通知");
}
/**
* 已传输完成的事件通知
* 1.QOS == 0,发送消息后会即可进行此事件回调，因为不需要等待回执
* 2.QOS == 1，发送消息后会等待ACK回执，ACK回执后会进行此事件通知
* 3.QOS == 2，发送消息后会等待PubRECV回执，知道收到PubCOMP后会进行此事件通知
* @param mqttMessageDeliveredEvent
*/
@EventListener(classes = MqttMessageDeliveredEvent.class)
public void listenerAction(MqttMessageDeliveredEvent mqttMessageDeliveredEvent) {
log.info("已传输完成的事件通知");
}
/**
* 消息订阅的事件通知
* @param mqttSubscribedEvent
*/
@EventListener(classes = MqttSubscribedEvent.class)
public void listenerAction(MqttSubscribedEvent mqttSubscribedEvent) {
log.info("消息订阅的事件通知");
}
}7、接口测试
@Resource
private MqttGateway mqttGateway;
/**
* sendData 消息
* topic 订阅主题
**/
@RequestMapping(value = "/sendMqtt",method = RequestMethod.POST)
public String sendMqtt(String sendData, String topic) {
MqttMessage mqttMessage = new MqttMessage();
mqttGateway.sendToMqtt(topic, sendData);
//mqttGateway.sendToMqttObject(topic, sendData.getBytes());
return "OK";
} 
本文提供Android接入MQTT服务并控制ESP8266的完整方案。通过规划主题、配置Android依赖与权限、订阅温湿度数据及下发LED控制指令，实现双向通信。适用于具备基础编程知识的开发者，帮助快速搭建物联网联调环境，解决连接失败、数据不同步等常见问题。
频繁断连问题可通过优化MQTT客户端配置提高稳定性。1.设置合理的KeepAlive时间（30-60秒），确保客户端定期发送PINGREQ包；2.配置遗嘱消息，使Broker在客户端异常掉线时发布通知；3.根据需求选择合适的QoS等级（0、1或2）以平衡可靠性与性能；4.启用自动重连机制并设置最大重连间隔；5.检查并优化网络环境，确保稳定连接；6.核对MQTTBroker配置，如最大连接数和socket超时时间；7.确保客户端ID唯一，避免冲突；8.完善异常处理机制，记录日志便于排查问题；9.优化订阅主题
如何使用PHP和MQTT为智能医疗设备添加实时监测功能导言：随着科技的不断发展，智能医疗设备在医疗行业中得到广泛应用。这些设备可以采集和传输患者的生理数据，例如心率、血压等，以实现实时监测和远程医疗。本文将介绍如何使用PHP和MQTT协议为智能医疗设备添加实时监测功能，并提供相关的代码示例。一、什么是MQTT协议？MQTT（MessageQueuingT
使用PHP和MQTT实现远程控制自动化生产线的数据通信在现代自动化生产线中，数据通信是非常重要的一环，它能够实现设备之间的实时数据传输和远程控制。而MQTT（MessageQueuingTelemetryTransport）是一种轻量级的、基于发布-订阅模式的即时消息传递协议，具有高效、可靠和灵活的特点。结合PHP语言的强大功能，我们可以很容易地实现远
PHP与MQTT:实现智能电力监控与管理系统的数据传输概述：随着智能电力监控与管理系统的普及，数据的传输变得至关重要。为了实现高效可靠的数据传输，一种适用于物联网的轻量级通信协议MQTT（MessageQueuingTelemetryTransport）应运而生。本文将介绍如何使用PHP语言结合MQTT协议实现智能电力监控与管理系统的数据传输。MQT
随着环境污染问题日益严重，环保监测系统成为保障人民健康、促进可持续发展的重要工具。在实现实时监测的过程中，PHP和MQTT技术的结合被广泛应用。一、MQTT技术介绍MQTT是一种发布-订阅式的消息传输协议，是轻量级的通信协议。通过MQTT可以实现高效、可扩展、可靠的异步通信，它适用于移动设备、低带宽、低计算能力的场景。MQTT协议有三个核心部分：代理、主题和
MQTT在PHP开发中的容错与安全性考虑概述：MQTT（MessageQueuingTelemetryTransport）是一种轻量级的通信协议，被广泛用于物联网和机器对机器（M2M）通信。在PHP开发中使用MQTT可以实现实时消息传递和远程控制等功能。本文将介绍在PHP开发中使用MQTT时需要考虑的容错和安全性问题，并提供一些代码示例供参考。一、容错
springBoot项目常用目录springBoot项目的目录结构及名规范介绍基于SpringBoot开发时的目录结构及命名规范，通过介绍能够为你解决，在实际项目中如何规划目录结构？如何命名目录更规范？各个目录分别代表什么含义？等三个问题。目录说明servicex//项目名|-admin-ui//管理服务前端代码(一般将UI和SERVICE放到一个工程中，便于管理)|-servicex-auth//模块1|-servicex-common//模块2|-servicex-gateway//模块3|-serv
PHP与MQTT实现实时数据传输的最佳实践引言：随着物联网技术的推进，实时数据传输变得越来越重要。为了实现实时数据传输，许多开发人员选择使用MQTT（MessageQueuingTelemetryTransport）协议。而PHP作为一种强大的开发语言，也成为了实现实时数据传输的首选。本文将介绍如何使用PHP与MQTT协议实现实时数据传输，并提供一些代
PHP和MQTT：构建基于消息队列的任务调度系统在当今互联网时代，随着互联网应用的快速发展，越来越多的任务需要被异步执行，以提高用户体验和系统性能。而传统的同步执行方式往往不再适用，这时候，消息队列就是一个非常好的选择。MQTT是一种轻量级的消息传输协议，它具有低耗能、低带宽占用、支持长连接等优势，使其成为构建基于消息队列的任务调度系统的理想选择。本文将介绍
火山引擎是字节跳动旗下企业级云与AI服务平台。聚焦大模型核心能力，依托字节自研豆包多模态大模型，凭借火山方舟一站式平台，提供模型精调、推理部署、智能应用搭建等全链路服务，结合高性能AI算力，助力各行业快速落地大模型数字化与智能化应用。
