Springboot对接mqtt的项目实践
更新时间：2026年02月10日 09:50:27 作者：IT界Tony哥
在SpringBoot中对接MQTT协议,可以使用Eclipse Paho客户端和Spring Integration MQTT模块,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
在Spring Boot中对接MQTT协议，可以使用Eclipse Paho客户端和Spring Integration MQTT模块。以下是详细实现步骤：
1. 添加依赖
<dependencies>
<!-- Spring Boot Starter -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- Spring Integration MQTT -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
<!-- Eclipse Paho MQTT Client -->
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
</dependencies>
2. 配置MQTT连接参数
# application.yml mqtt: broker-url: tcp://localhost:1883 username: admin password: password client-id: spring-boot-client default-topic: test/topic timeout: 30 keepalive: 60 completion-timeout: 30000
3. MQTT配置类
@Configuration
@EnableConfigurationProperties(MqttProperties.class)
public class MqttConfig {
@Autowired
private MqttProperties mqttProperties;
// MQTT连接配置
@Bean
public MqttConnectOptions mqttConnectOptions() {
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{mqttProperties.getBrokerUrl()});
options.setUserName(mqttProperties.getUsername());
options.setPassword(mqttProperties.getPassword().toCharArray());
options.setConnectionTimeout(mqttProperties.getTimeout());
options.setKeepAliveInterval(mqttProperties.getKeepalive());
options.setAutomaticReconnect(true);
options.setCleanSession(true);
return options;
}
// MQTT客户端工厂
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(mqttConnectOptions());
return factory;
}
// 出站消息通道（用于发送消息）
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler =
new MqttPahoMessageHandler(mqttProperties.getClientId() + "-producer", mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttProperties.getDefaultTopic());
return messageHandler;
}
// 出站通道
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
// 入站消息适配器（用于接收消息）
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "-consumer",
mqttClientFactory(), mqttProperties.getDefaultTopic());
adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInboundChannel());
return adapter;
}
// 入站通道
@Bean
public MessageChannel mqttInboundChannel() {
return new DirectChannel();
}
// 入站消息处理器
@Bean
@ServiceActivator(inputChannel = "mqttInboundChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String topic = (String) message.getHeaders().get("mqtt_receivedTopic");
String payload = (String) message.getPayload();
System.out.println("Received message from topic: " + topic + ", payload: " + payload);
// 处理接收到的消息
processMessage(topic, payload);
}
};
}
}
4. 配置属性类
@ConfigurationProperties(prefix = "mqtt")
@Component
@Data
public class MqttProperties {
private String brokerUrl;
private String username;
private String password;
private String clientId;
private String defaultTopic;
private int timeout;
private int keepalive;
private int completionTimeout;
}
5. MQTT服务类
@Service
public class MqttService {
@Autowired
private MessageChannel mqttOutboundChannel;
// 发送消息到指定主题
public void sendMessage(String topic, String message) {
mqttOutboundChannel.send(MessageBuilder.withPayload(message)
.setHeader("mqtt_topic", topic)
.build());
}
// 发送消息到默认主题
public void sendMessage(String message) {
mqttOutboundChannel.send(MessageBuilder.withPayload(message).build());
}
// 发送带QoS的消息
public void sendMessage(String topic, String message, int qos) {
mqttOutboundChannel.send(MessageBuilder.withPayload(message)
.setHeader("mqtt_topic", topic)
.setHeader("mqtt_qos", qos)
.build());
}
}
6. 消息处理器
@Component
public class MqttMessageProcessor {
private static final Logger logger = LoggerFactory.getLogger(MqttMessageProcessor.class);
public void processMessage(String topic, String payload) {
logger.info("Processing MQTT message - Topic: {}, Payload: {}", topic, payload);
// 根据不同的主题进行不同的处理
switch (topic) {
case "test/topic":
handleTestTopic(payload);
break;
case "sensor/data":
handleSensorData(payload);
break;
default:
handleDefaultMessage(topic, payload);
}
}
private void handleTestTopic(String payload) {
logger.info("处理测试主题消息: {}", payload);
// 具体的业务逻辑
}
private void handleSensorData(String payload) {
logger.info("处理传感器数据: {}", payload);
try {
// 解析JSON数据等操作
// ObjectMapper mapper = new ObjectMapper();
// SensorData data = mapper.readValue(payload, SensorData.class);
} catch (Exception e) {
logger.error("解析传感器数据失败", e);
}
}
private void handleDefaultMessage(String topic, String payload) {
logger.info("处理默认消息 - Topic: {}, Payload: {}", topic, payload);
}
}
7. 控制器示例
@RestController
@RequestMapping("/mqtt")
public class MqttController {
@Autowired
private MqttService mqttService;
@PostMapping("/publish")
public ResponseEntity<String> publishMessage(@RequestParam String topic,
@RequestParam String message) {
try {
mqttService.sendMessage(topic, message);
return ResponseEntity.ok("Message published successfully");
} catch (Exception e) {
return ResponseEntity.status(500).body("Failed to publish message: " + e.getMessage());
}
}
@PostMapping("/publish/default")
public ResponseEntity<String> publishToDefaultTopic(@RequestParam String message) {
try {
mqttService.sendMessage(message);
return ResponseEntity.ok("Message published to default topic");
} catch (Exception e) {
return ResponseEntity.status(500).body("Failed to publish message: " + e.getMessage());
}
}
}
8. 主应用类
@SpringBootApplication
@EnableConfigurationProperties
public class MqttApplication {
public static void main(String[] args) {
SpringApplication.run(MqttApplication.class, args);
}
}
9. 测试MQTT服务
可以使用MQTT.fx或其他MQTT客户端工具进行测试：
- 启动Spring Boot应用
- 使用MQTT客户端订阅主题
test/topic - 调用API发送消息：
curl -X POST "http://localhost:8080/mqtt/publish?topic=test/topic&message=Hello MQTT"
主要特性
- 自动重连: 配置了自动重连机制
- QoS支持: 支持不同的服务质量等级
- 多主题订阅: 可以订阅多个主题
- 异步处理: 消息发送支持异步模式
- 配置灵活: 通过配置文件管理连接参数
这样你就实现了一个完整的Spring Boot MQTT集成方案，可以方便地进行消息的发布和订阅。
到此这篇关于Springboot对接mqtt的项目实践的文章就介绍到这了,更多相关Springboot对接mqtt内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了SpringBoot @NotBlank错误的解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-08-08
这篇文章主要介绍了分享JVM 的四种引用方式，文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下2022-07-07
这篇文章主要介绍了java中带参数的try(){}语法含义详解，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2022-02-02
这篇文章主要给大家介绍了关于Spring Cloud Zuul过滤器的相关资料,通过阅读本文您将了解：Zuul过滤器类型与请求生命周期、如何编写Zuul过滤器、如何禁用Zuul过滤器和Spring Cloud为Zuul编写的过滤器及其功能，需要的朋友可以参考下。2017-02-02
这篇文章主要介绍了spring AOP的After增强实现方法,结合实例形式分析了spring面向切面AOP的After增强实现步骤与相关操作技巧,需要的朋友可以参考下2020-01-01
这篇文章主要为大家详细介绍了java实现单链表中的增删改，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2022-05-05
本文将深入探讨如何使用Java的EnTab和DeTab类实现制表符与空格的智能转换,并分析其在实际开发中的应用,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2025-04-04
顺序表是计算机内存中以数组的形式保存的线性表，线性表的顺序存储是指用一组地址连续的存储单元依次存储线性表中的各个元素、使得线性表中在逻辑结构上相邻的数据元素存储在相邻的物理存储单元中，即通过数据元素物理存储的相邻关系来反映数据元素之间逻辑上的相邻关系2021-10-10
本文主要介绍了servlet创建web后端程序的示例代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2023-06-06
这篇文章主要介绍了Java工程如何打印程序日志过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-10-10
最新评论