MQTT：Java集成MQTT

Git項目路徑
一、原生java架構
- 1.1 導入POM文件
- 1.2 編寫測試用例
二、SpringBoot集成MQTT
- 2.1 導入POM文件
- 2.2 在YML文件中增加配置
- 2.3 新建Properties配置文件映射配置
- 2.4 創建連接工廠
- 2.5 增加入站規則配置
- 2.6 增加出站規則配置
- 2.7 創建消息發送網關
- 2.8 測試消息發送
- 2.9 項目結構

Git項目路徑

一、原生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 ...");}
}