MQTT：Java集成MQTT

---

一、原生java架构

1.1 导入POM文件

<dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.5</version>
</dependency>

1.2 编写测试用例

package com.ming;import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;/*** 使用java原生方法连接MQTT*/
public class MqttPahoTest {private final String serverURI = "tcp://localhost:1883";private final String clientId = "emqx_spring_client_132";/*** 建立连接* @throws MqttException*/@Testpublic MqttClient createConnection() throws MqttException {// 创建MQTT对象MqttClient client = new MqttClient(serverURI, clientId, new MemoryPersistence());// 发送建立连接的请求MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();mqttConnectOptions.setUserName("admin");mqttConnectOptions.setPassword("admin".toCharArray());mqttConnectOptions.setCleanSession(true);client.connect(mqttConnectOptions);return client;}/*** 发送消息* @throws MqttException*/@Testpublic void sendMsg() throws MqttException {// 创建对象MqttClient client = createConnection();// 发送消息MqttMessage mqttMessage = new MqttMessage();mqttMessage.setQos(2);mqttMessage.setPayload("Hello World".getBytes());client.publish("java/a", mqttMessage);// 关闭连接client.disconnect();client.close();}/*** 接收消息* @throws MqttException*/@Testpublic void receiveMsg() throws MqttException {// 创建MQTT对象MqttClient client = new MqttClient(serverURI, clientId, new MemoryPersistence());// 发送建立连接的请求MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();mqttConnectOptions.setUserName("admin");mqttConnectOptions.setPassword("admin".toCharArray());mqttConnectOptions.setCleanSession(true);client.setCallback(new MqttCallback() {@Overridepublic void connectionLost(Throwable throwable) {  // 当连接丢失时的回调System.out.println("Connection lost...");}@Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {  // 消息接收回调System.out.println(String.format("%s ---> %s", topic, new String(mqttMessage.getPayload())));}@Overridepublic void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {  // 消息传输完毕System.out.println("Delivery complete");}});client.connect(mqttConnectOptions);// 订阅主题client.subscribe("java/b", 2);while (true);}
}

二、SpringBoot集成MQTT

2.1 导入POM文件

<!-- spring boot 项目集成消息中间件基础依赖 -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<!-- spring boot 项目和MQTT客户端集成依赖 -->
<dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId><version>5.4.3</version>
</dependency>
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId>
</dependency>
<dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.83</version>
</dependency>

2.2 在YML文件中增加配置

spring:mqtt:username: adminpassword: adminurl: tcp://localhost:1883subClientId: sub_client_id_123subTopic: atguigu/iot/lamp/line1,atguigu/iot/lamp/line2pubClientId: pub_client_id_123

2.3 新建Properties配置文件映射配置

package com.ming.properties;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;@Data
@ConfigurationProperties(prefix = "spring.mqtt")
public class MqttConfigurationProperties {private String username;private String password;private String url;private String subClientId;private String subTopic;private String pubClientId;
}

2.4 创建连接工厂

package com.ming.config;import com.ming.properties.MqttConfigurationProperties;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;/*** 创建MQTT的配置类 配置连接工厂*/
@Configuration
public class MqttConfiguration {@Autowiredprivate MqttConfigurationProperties mqttConfigurationProperties;@Beanpublic MqttPahoClientFactory mqttPahoClientFactory() {DefaultMqttPahoClientFactory mqttPahoClientFactory = new DefaultMqttPahoClientFactory();MqttConnectOptions options = new MqttConnectOptions();options.setUserName(mqttConfigurationProperties.getUsername());options.setPassword(mqttConfigurationProperties.getPassword().toCharArray());options.setServerURIs(new String[]{mqttConfigurationProperties.getUrl()});mqttPahoClientFactory.setConnectionOptions(options);return mqttPahoClientFactory;}
}

2.5 增加入站规则配置

package com.ming.config;import com.ming.handler.ReceiverMessageHandler;
import com.ming.properties.MqttConfigurationProperties;
import org.springframework.beans.factory.annotation.Autowired;
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.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;/*** MQTT入站规则配置类（接收消息）*/
@Configuration
public class MqttInboundConfiguration {@Autowiredprivate MqttConfigurationProperties mqttConfigurationProperties;@Autowiredprivate MqttPahoClientFactory mqttPahoClientFactory;@Autowiredprivate ReceiverMessageHandler receiverMessageHandler;/*** 消息通道* @return*/@Beanpublic MessageChannel messageInboundChannel() {return new DirectChannel();}/*** 配置入站适配器，作用：设置订阅主题，以及指定消息的相关属性* @return*/@Beanpublic MessageProducer messageProducer() {MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfigurationProperties.getUrl(),mqttConfigurationProperties.getSubClientId(),mqttPahoClientFactory,mqttConfigurationProperties.getSubTopic().split(","));mqttPahoMessageDrivenChannelAdapter.setQos(1);mqttPahoMessageDrivenChannelAdapter.setConverter(new DefaultPahoMessageConverter());mqttPahoMessageDrivenChannelAdapter.setOutputChannel(messageInboundChannel());return mqttPahoMessageDrivenChannelAdapter;}/*** 消息处理器* @return*/@Bean@ServiceActivator(inputChannel = "messageInboundChannel")public MessageHandler messageHandler() {return receiverMessageHandler;}
}

package com.ming.handler;import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;/*** 当订阅的主题有消息时就会触发此处回调*/
@Component
public class ReceiverMessageHandler implements MessageHandler {@Overridepublic void handleMessage(Message<?> message) throws MessagingException {Object payload = message.getPayload();  // 获取消息的内容System.out.println(message.getHeaders().get("mqtt_receivedTopic"));  // 主题名称System.out.println(payload);  // 消息主体}
}

2.6 增加出站规则配置

package com.ming.config;import com.ming.properties.MqttConfigurationProperties;
import org.springframework.beans.factory.annotation.Autowired;
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.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;/*** MQTT出站规则配置类（发送消息）*/
@Configuration
public class MqttOutboundConfiguration {@Autowiredprivate MqttConfigurationProperties mqttConfigurationProperties;@Autowiredprivate MqttPahoClientFactory mqttPahoClientFactory;/*** 消息通道* @return*/@Beanpublic MessageChannel mqttOutboundChannel() {return new DirectChannel();}/*** 配置出站适配器* @return*/@Bean@ServiceActivator(inputChannel = "mqttOutboundChannel")public MessageHandler mqttOutboundMessageHandler() {MqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler(mqttConfigurationProperties.getUrl(),mqttConfigurationProperties.getPubClientId(),mqttPahoClientFactory);mqttPahoMessageHandler.setDefaultQos(0);mqttPahoMessageHandler.setDefaultTopic("default");mqttPahoMessageHandler.setAsync(true);return mqttPahoMessageHandler;}
}

2.7 创建消息发送网关

package com.ming.getway;import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;/*** MQTT发送消息的网关*/
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGetWay {public abstract void sendMsgToMqtt(@Header(value = MqttHeaders.TOPIC) String topic, String payload);public abstract void sendMsgToMqtt(@Header(value = MqttHeaders.TOPIC) String topic, @Header(value = MqttHeaders.QOS) int qos,String payload);
}

package com.ming.service;import com.ming.getway.MqttGetWay;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class MqttMessageSender {@Autowiredprivate MqttGetWay mqttGetWay;public void sendMsg(String topic, String message) {mqttGetWay.sendMsgToMqtt(topic, message);}public void sendMsg(String topic, int qos, String message) {mqttGetWay.sendMsgToMqtt(topic, qos, message);}
}

2.8 测试消息发送

package com.ming;import com.ming.service.MqttMessageSender;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest(classes = SpringMqttDemoApplication.class)
public class MqttMessageSenderTest {@Autowiredprivate MqttMessageSender mqttMessageSender;@Testpublic void sendToMsg(){mqttMessageSender.sendMsg("java/c","hello mqtt spring boot ...");}
}

2.9 项目结构