最近在做MQTT对接，然后发送消息，然后参考网上的实战文章进行了一下整理。
文章主要参考自（https://www.codetd.com/article/13550340），然后自己做了些许更改。
1、整合准备SpringBoot：2.2.2.RELEASE
MQTT平台：EMQX4.4.1（Docker运行）
虚拟机服务器：Centos7（192.168.56.102 ）
发送端：cloud-mqtt-send8001
接收端：cloud-mqtt-accept8002
2、发送端：cloud-mqtt-send8002导入POM依赖：
org.springframework.boot
spring-boot-starter-integration
org.springframework.integration
spring-integration-stream
org.springframework.integration
spring-integration-mqtt
org.springframework.boot
spring-boot-configuration-processor
true
org.projectlombok
lombok
1.18.22
provided
设置YML
server:
port: 8001
spring:
application:
name: mqtt-send
#mqtt属性配置
mqtt:
hostUrl: tcp://192.168.56.102:1883
username: admin
password: public
clientid: mqtt_send_client
cleanSession: true
reconnect: true
#连接超时
timeout: 1000
#设置会话心跳时间
keepalive: 100
defaultTopic: client:report:1
isOpen: true
qos: 1
新建配置类：MqttProperties.java
@Data
@Component
@ConfigurationProperties(prefix = "mqtt")
public class MqttProperties {
private String username;
private String password;
private String hostUrl;
private String clientId;
private String defaultTopic;
private int timeout;
private int keepAlive;
private Boolean cleanSession;
private Boolean reconnect;
private Boolean isOpen;
private Integer qos;
}
添加MQTT发送客户端：MqttSendClient：
@Slf4j
@Component
public class MqttSendClient {
@Autowired
private MqttSendCallBack mqttSendCallBack;
@Autowired
private MqttProperties mqttProperties;
private static MqttClient mqttClient;
private static MqttClient getClient() {
return mqttClient;
}
private static void setClient(MqttClient client) {
MqttSendClient.mqttClient = client;
}
public void connect(){
MqttClient client = null;
try {
//String uuid = UUID.randomUUID().toString().replaceAll("-",""); //设置每一个客户端的id
client = new MqttClient(mqttProperties.getHostUrl(),mqttProperties.getClientId() , new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(mqttProperties.getUsername());
options.setPassword(mqttProperties.getPassword().toCharArray());
options.setConnectionTimeout(mqttProperties.getTimeout());
options.setKeepAliveInterval(mqttProperties.getKeepAlive());
options.setCleanSession(true);
options.setAutomaticReconnect(false);
MqttSendClient.setClient(client);
try {
// 设置回调
client.setCallback(mqttSendCallBack);
client.connect(options);
} catch (Exception e) {
e.printStackTrace();
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void publish(String topic, String pushMessage) {
publish(0, false, topic, pushMessage);
}
public void publish(int qos, boolean retained, String topic, String pushMessage) {
MqttMessage message = new MqttMessage();
message.setQos(qos);
message.setRetained(retained);
message.setPayload(pushMessage.getBytes());
MqttTopic mTopic = MqttSendClient.getClient().getTopic(topic);
if (null == mTopic) {
log.error("主题不存在:{}",mTopic);
}
try {
mTopic.publish(message);
log.info("消息发送成功");
} catch (Exception e) {
log.error("mqtt发送消息异常:",e);
}
}
}
添加MQTT发送客户端回调类：MqttSendCallBack
@Slf4j
@Component
public class MqttSendCallBack implements MqttCallbackExtended {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
log.info("————————————————-ClientID:{}——————————————"+"链接成功");
}
@Override
public void connectionLost(Throwable cause) {
log.error("【MQTT-发送端】链接断开！");
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
log.info("【MQTT-发送端】接收消息主题 : " + topic);
log.info("【MQTT-发送端】接收消息Qos : " + message.getQos());
log.info("【MQTT-发送端】接收消息内容 : " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
String[] topics = token.getTopics();
if (topics!=null && topics.length>0){
for (String topic : topics) {
log.info("【MQTT-发送端】向主题：" + topic + "发送消息成功！");
}
}
try {
MqttMessage message = token.getMessage();
byte[] payload = message.getPayload();
String s = new String(payload, "UTF-8");
log.info("【MQTT-发送端】消息的内容是：" + s);
} catch (MqttException e) {
e.printStackTrace();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
添加：MqttCondition
public class MqttCondition implements Condition {
@Override
public boolean matches(ConditionContext context, AnnotatedTypemetadata metadata) {
System.out.println("MqttCondition。。。。");
//1、能获取到ioc使用的beanfactory
ConfigurableListableBeanFactory beanFactory = context.getBeanFactory();
//2、获取类加载器
ClassLoader classLoader = context.getClassLoader();
//3、获取当前环境信息
Environment environment = context.getEnvironment();
String isOpen = environment.getProperty("mqtt.isOpen");
return Boolean.valueOf(isOpen);
}
}
添加MQTT配置类：MqttConfig
@Configuration
public class MqttConfig {
@Autowired
private MqttSendClient mqttSendClient;
@Conditional(MqttCondition.class)
@Bean
public MqttSendClient getMqttSendClient(){
mqttSendClient.connect();
return mqttSendClient;
}
}
主启动类
@SpringBootApplication
public class MqttSendApplication {
public static void main(String[] args) {
SpringApplication.run(MqttSendApplication.class, args);
}
}
启动项目，链接MQTT服务器成功。
项目整体代码结构如下：
3、接收端：cloud-mqtt-accept8002
导入POM
org.springframework.boot spring-boot-starter-weborg.springframework.boot spring-boot-starter-actuatororg.springframework.boot spring-boot-devtoolsruntime true org.projectlombok lomboktrue org.springframework.boot spring-boot-starter-testtest org.eclipse.paho org.eclipse.paho.client.mqttv31.2.2 org.springframework.boot spring-boot-configuration-processortrue
设置YML
server:
port: 8002
spring:
application:
name: mqtt-accept
mqtt:
hostUrl: tcp://192.168.56.102:1883
username: admin
password: public
clientid: mqtt_accept_client
cleanSession: true
reconnect: true
#连接超时
timeout: 1000
#设置会话心跳时间
keepalive: 100
defaultTopic: client:report:1
isOpen: true
qos: 1
属性配置文件：MqttProperties.java
@Data
@Component
@ConfigurationProperties(prefix = "mqtt")
public class MqttProperties {
private String username;
private String password;
private String hostUrl;
private String clientId;
private String defaultTopic;
private int timeout;
private int keepAlive;
private Boolean cleanSession;
private Boolean reconnect;
private Boolean isOpen;
private Integer qos;
}
添加MQTT接收客户端：MqttAcceptClient
@Slf4j
@Component
public class MqttAcceptClient {
@Autowired
private MqttAcceptCallback mqttAcceptCallback;
@Autowired
private MqttProperties mqttProperties;
private static MqttClient mqttClient;
public static MqttClient getMqttClient() {
return mqttClient;
}
public static void setMqttClient(MqttClient mqttClient) {
MqttAcceptClient.mqttClient = mqttClient;
}
public void connect() {
MqttClient client;
try {
client = new MqttClient(mqttProperties.getHostUrl(), mqttProperties.getClientId(), new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(mqttProperties.getUsername());
options.setPassword(mqttProperties.getPassword().toCharArray());
options.setConnectionTimeout(mqttProperties.getTimeout());
options.setKeepAliveInterval(mqttProperties.getKeepAlive());
options.setAutomaticReconnect(mqttProperties.getReconnect());
options.setCleanSession(mqttProperties.getCleanSession());
MqttAcceptClient.setMqttClient(client);
try {
// 设置回调
client.setCallback(mqttAcceptCallback);
client.connect(options);
} catch (Exception e) {
e.printStackTrace();
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void reconnection() {
try {
mqttClient.connect();
} catch (MqttException e) {
e.printStackTrace();
}
}
public void subscribe(String topic, int qos) {
log.info("==============开始订阅主题==============" + topic);
try {
mqttClient.subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
public void unsubscribe(String topic) {
log.info("==============开始取消订阅主题==============" + topic);
try {
mqttClient.unsubscribe(topic);
} catch (MqttException e) {
e.printStackTrace();
}
}
}
添加mqtt接受服务的回调类:MqttAcceptCallback
@Slf4j
@Component
public class MqttAcceptCallback implements MqttCallbackExtended {
@Autowired
private MqttAcceptClient mqttAcceptClient;
@Override
public void connectionLost(Throwable throwable) {
log.info("【MQTT-消费端】连接断开，可以做重连");
if (MqttAcceptClient.getMqttClient() == null || !MqttAcceptClient.getMqttClient().isConnected()) {
log.info("【MQTT-消费端】emqx重新连接....................................................");
mqttAcceptClient.reconnection();
}
}
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
log.info("【MQTT-消费端】接收消息主题 : " + topic);
log.info("【MQTT-消费端】接收消息Qos : " + mqttMessage.getQos());
log.info("【MQTT-消费端】接收消息内容 : " + new String(mqttMessage.getPayload()));
// int i = 1/0;
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
String[] topics = token.getTopics();
for (String topic : topics) {
log.info("【MQTT-消费端】向主题：" + topic + "发送消息成功！");
}
try {
MqttMessage message = token.getMessage();
byte[] payload = message.getPayload();
String s = new String(payload, "UTF-8");
log.info("【MQTT-消费端】消息的内容是：" + s);
} catch (MqttException e) {
e.printStackTrace();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void connectComplete(boolean b, String s) {
System.out.println("s: " + s);
log.info("--------------------【MQTT-消费端】连接成功！--------------------");
// 以/#结尾表示订阅所有以test开头的主题
// 订阅所有机构主题
mqttAcceptClient.subscribe("test_queue", 0);
}
}
MqttCondition
public class MqttCondition implements Condition {
@Override
public boolean matches(ConditionContext context, AnnotatedTypemetadata metadata) {
System.out.println("MqttCondition。。。。");
//1、能获取到ioc使用的beanfactory
ConfigurableListableBeanFactory beanFactory = context.getBeanFactory();
//2、获取类加载器
ClassLoader classLoader = context.getClassLoader();
//3、获取当前环境信息
Environment environment = context.getEnvironment();
String isOpen = environment.getProperty("mqtt.isOpen");
return Boolean.valueOf(isOpen);
}
}
MQTT启动配置类：MqttConfig.java
@Configuration
public class MqttConfig {
@Autowired
private MqttAcceptClient mqttAcceptClient;
@Conditional(MqttCondition.class)
@Bean
public MqttAcceptClient getMqttAcceptClient(){
mqttAcceptClient.connect();
//mqttAcceptClient.subscribe("test_queue",0);
return mqttAcceptClient;
}
}
主启动类
package com.xlh.springcloud;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class MqttAcceptApplication {
public static void main(String[] args) {
SpringApplication.run(MqttAcceptApplication.class,args);
}
}
启动项目，链接MQTT服务器，然后订阅主题：
项目整体结构如下：
4、项目测试
分别启动发送端和接收端以后，EMQX监控平台如下：
然后调用发送端测试接口：http://localhost:8001/send
发送端cloud-mqtt-send8001运行结果如下：
接收端cloud-mqtt-accept8002运行结果如下：
可以看到接收端成功的接收到消息
