Spring Boot使用MQTT与硬件设备进行交互

Spring Boot需要与硬件设备进行交互，使用MQTT作为通信协议。对于这种情况，MQTT的服务端（Broker）需要在服务器上安装和配置，硬件设备和Java应用程序（Spring Boot）通过该Broker进行消息的发布和订阅。以下是详细步骤：

一、服务器安装步骤

安装 Mosquitto (MQTT Broker)

Ubuntu:

# 更新软件包列表
sudo apt-get update
# 安装 Mosquitto 及其客户端工具
sudo apt-get install mosquitto mosquitto-clients
# 配置 Mosquitto 开机自启动
sudo systemctl enable mosquitto
# 启动 Mosquitto 服务
sudo systemctl start mosquitto

CentOS:

# 更新软件包列表
sudo yum update
# 安装 EPEL 仓库
sudo yum install epel-release
# 安装 Mosquitto
sudo yum install mosquitto
# 配置 Mosquitto 开机自启动
sudo systemctl enable mosquitto
# 启动 Mosquitto 服务
sudo systemctl start mosquitto

配置 Mosquitto（可选）

默认情况下，Mosquitto 允许匿名连接，但为了安全性，可以配置用户名和密码。在 /etc/mosquitto/mosquitto.conf 文件中进行配置：

# 启用监听1883端口
listener 1883
# 允许匿名访问
allow_anonymous true

二、Spring Boot 配置和工具类

添加依赖

在pom.xml文件中添加以下依赖：

<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>
</dependency>

application.yml配置

可以将所有MQTT相关的配置，包括client_id和默认主题，配置在application.yml中

mqtt:broker:# 设备MQTT应用配置url: tcp://your_server_ip:1883username: your_usernamepassword: your_password# 自定义约定client_idclientId: your_client_idtopics:# 默认监听default: your/default_topic# 自定义监听subscriptions:- your/topic1- your/topic2

配置类

创建一个 MqttConfig 配置类：

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
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.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.beans.factory.annotation.Autowired;@Configuration
public class MqttConfig {@Autowiredprivate MqttUtil mqttUtil;@Value("${mqtt.broker.url}")private String brokerUrl;@Value("${mqtt.broker.username}")private String username;@Value("${mqtt.broker.password}")private String password;@Value("${mqtt.broker.clientId}")private String clientId;@Value("${mqtt.topics.default}")private String defaultTopic;@Value("${mqtt.topics.subscriptions}")private String[] topics;// 配置MQTT客户端工厂@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();MqttConnectOptions options = new MqttConnectOptions();options.setServerURIs(new String[]{brokerUrl});options.setUserName(username);options.setPassword(password.toCharArray());factory.setConnectionOptions(options);return factory;}// （接收消息:步骤一）定义一个消息通道，用于接收消息@Beanpublic MessageChannel mqttInputChannel() {return new DirectChannel();}// （接收消息:步骤二）配置消息驱动通道适配器，用于接收指定主题的消息@Beanpublic MessageProducer inbound() {MqttPahoMessageDrivenChannelAdapter adapter =new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(), topics);adapter.setOutputChannel(mqttInputChannel());return adapter;}// （接收消息:步骤三）配置消息处理器，用于处理接收到的消息@Bean@ServiceActivator(inputChannel = "mqttInputChannel")public MessageHandler handler() {return message -> {String topic = message.getHeaders().get("mqtt_receivedTopic").toString();String payload = message.getPayload().toString();System.out.println("接收到消息: 主题 = " + topic + ", 内容 = " + payload);// 发布接收到的消息事件mqttUtil.publishReceivedMessage(topic, payload);};}// 配置MQTT消息发送处理器@Bean@ServiceActivator(inputChannel = "mqttOutboundChannel")public MessageHandler mqttOutbound() {MqttPahoMessageHandler messageHandler =new MqttPahoMessageHandler(clientId, mqttClientFactory());messageHandler.setAsync(true);messageHandler.setDefaultTopic(defaultTopic); // 设置默认主题return messageHandler;}// 定义一个消息通道，用于发送消息@Beanpublic MessageChannel mqttOutboundChannel() {return new DirectChannel();}
}

MQTT 工具类
创建一个 MqttUtil 工具类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;@Component
public class MqttUtil {@Autowiredprivate MessageChannel mqttOutboundChannel;@Autowiredprivate ApplicationEventPublisher eventPublisher;/*** 发送消息到默认主题* @param payload 消息内容*/public void sendMessage(String payload) {mqttOutboundChannel.send(MessageBuilder.withPayload(payload).build());}/*** 发送消息到指定主题* @param topic 主题* @param payload 消息内容*/public void sendMessage(String topic, String payload) {mqttOutboundChannel.send(MessageBuilder.withPayload(payload).setHeader("mqtt_topic", topic).build());}/*** （接收消息:步骤四）发布接收到的MQTT消息事件* @param topic 主题* @param payload 消息内容*/public void publishReceivedMessage(String topic, String payload) {MqttMessageEvent event = new MqttMessageEvent(this, topic, payload);eventPublisher.publishEvent(event);}
}

MQTT 消息事件( MqttUtil.publishReceivedMessage() 方法使用)

创建一个事件类，用于封装接收到的MQTT消息：

import org.springframework.context.ApplicationEvent;public class MqttMessageEvent extends ApplicationEvent {private final String topic;private final String payload;public MqttMessageEvent(Object source, String topic, String payload) {super(source);this.topic = topic;this.payload = payload;}public String getTopic() {return topic;}public String getPayload() {return payload;}
}

消息监听器( MqttUtil.publishReceivedMessage() 方法使用)

创建一个监听器，用于处理接收到的MQTT消息：

import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;@Component
public class MqttMessageListener {/** * （接收消息:步骤五）事件监听器中处理消息 *     说明：此方法将接收到上方 MqttUtil.publishReceivedMessage() 发布的事件并处理这些消息 * @param event 接收到的事件 */@EventListenerpublic void handleMqttMessage(MqttMessageEvent event) {String topic = event.getTopic();String payload = event.getPayload();System.out.println("处理接收到的消息: 主题 = " + topic + ", 内容 = " + payload);// 根据不同的主题进行不同的处理if ("your/topic1".equals(topic)) {handleTopic1Message(payload);} else if ("your/topic2".equals(topic)) {handleTopic2Message(payload);}}private void handleTopic1Message(String payload) {// 处理来自 topic1 的消息System.out.println("处理 topic1 消息: " + payload);}private void handleTopic2Message(String payload) {// 处理来自 topic2 的消息System.out.println("处理 topic2 消息: " + payload);}
}

使用示例

在你的Spring Boot应用中，可以通过注入MqttUtil来发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class MqttController {@Autowiredprivate MqttUtil mqttUtil;@GetMapping("/send")public String sendMessage() {mqttUtil.sendMessage("your/topic1", "Hello from Spring Boot");return "Message sent!";}
}