概念先行：

mqtt与mq的关系

MQTT（Message Queuing Telemetry Transport）是一种基于发布/订阅模式的通信协议，它与MQ（Message Queue，消息队列）有一定的关联，但二者并不完全相同。

MQTT是一种轻量级的通信协议，专门为在物联网（IoT）设备之间的消息传递而设计。它运行在TCP协议之上，以“发布-订阅”模式进行消息传递。在这种模式中，发布者将消息发布到特定的主题（topic）中，而订阅者则订阅这些主题以获取消息。

另一方面，MQ是一种更为通用的消息队列技术，它可以支持多种不同的消息传递协议，包括MQTT、AMQP（Advanced Message Queuing Protocol）、STOMP（Streaming Text Oriented Message Protocol）等。MQ可以提供更为复杂和灵活的消息传递模式，包括点对点、发布/订阅等。

尽管MQTT和MQ是不同的概念，但它们可以在实际应用中进行结合。例如，你可以使用MQTT协议将消息发送到MQ中，然后使用MQ的其他功能对这些消息进行处理和传递。

总之，MQTT是一种通信协议，而MQ是一种消息队列技术，二者虽然有关联，但概念不同。

EMQX在本次技术中担任什么角色？解决什么问题？

1、在技术中担任什么角色？

在物联网中，EMQX使用MQTT协议来担任消息传递的角色。具体来说，EMQX是一种MQTT broker的实现，它充当了MQTT网络中的服务器，负责接收和转发客户端发布的MQTT消息。

MQTT是一种轻量级的发布/订阅模式的通信协议，非常适合在物联网设备之间进行消息传递。通过使用MQTT协议，设备可以发布和订阅不同的主题，从而实现设备间的信息交流和数据传输。

在技术中，EMQX作为MQTT broker，它的主要职责是接收来自客户端的MQTT连接请求，处理并转发客户端发布的MQTT消息。同时，EMQX也负责处理客户端的订阅请求，将订阅的主题与相应的客户端进行关联，确保只有订阅了特定主题的客户端才能接收到相关的消息。

除了基本的消息传递功能，EMQX还提供了许多其他的高级功能，例如安全性和认证、QoS（Quality of Service）控制、持久化存储等。这些功能进一步增强了MQTT在物联网中的应用价值和实用性。

EMQX在物联网中担任的角色是MQTT broker，负责实现MQTT协议的消息传递功能，并提供额外的功能和特性以满足物联网应用的需求。

2、解决了什么问题？

在物联网中，使用EMQX和MQTT协议可以解决以下问题：

设备连接和通信：EMQX可以连接大量的物联网设备，并支持它们之间的通信。这使得设备可以相互传递信息，进行数据交换和协同工作。
数据采集和监控：通过EMQX，可以对大量设备进行数据采集和监控，实时获取设备的工作状态和运行数据。这有助于及时发现问题并进行处理，同时也可以进行远程监控和控制。
实时性通知：使用MQTT的发布/订阅模式，设备可以及时接收和响应其他设备发布的信息。这可以实现实时性的通知和提醒，例如在设备出现故障或异常时向管理员发送警报。
能耗优化：由于MQTT协议的轻量级特性，使用EMQX进行消息传递可以降低设备的能耗。这尤其适用于电池供电的物联网设备，可以延长其使用寿命。
数据持久化：EMQX支持将数据存储在本地或远程数据库中，实现数据的持久化存储。这有助于对历史数据进行查询和分析，支持决策和预测。

使用EMQX和MQTT协议可以解决物联网中的设备连接、数据采集、实时通知、能耗优化和数据持久化等问题，为物联网应用的开发和实施提供有力的支持。

因为使用了这个类似于mq的中间键，方便了服务器去获取信息和处理信息。

准备工作：

准备一台安装部署好的emqx服务器，搭建文章如下。

MQTT协议--技术文档--搭建mqtt服务器--《EMQX单体服务器部署》_一单成的博客-CSDN博客

开始代码demo演示

注意：文章中描述了两种角色

1、发布者

2、订阅者

发布者和订阅者都是根据主题来发送和存储消息的。他们两个并不知道彼此。并且根据实际的业务逻辑有的终端可以既是一些信息的发布者，也可以是一些主题的订阅者。用来拿取和接收信息。

订阅者代码demo演示：

订阅者代码结构展示：

配置文件展示：

pom文件依赖

  <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- mqtt --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-integration</artifactId></dependency><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-stream</artifactId></dependency><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId></dependency><!--配置文件报错问题--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency><!--lombok--><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.22</version><scope>provided</scope></dependency></dependencies>

application.yml文件配置展示

# 服务器相关server:port: 10002spring:main:allow-circular-references: trueapplication:name: dispatchermqtt:hostUrl: tcp://ip:1883username: devpassword: devclient-id: MQTT-CLIENT-DEVcleanSession: truereconnect: truetimeout: 100keepAlive: 100defaultTopic: server/dev/reportserverTopic: server/dev/reportisOpen: trueqos: 0

配置文件详解：

server: - 开始配置服务器相关的关键字
port: 10002 - 服务器监听的端口号。这里配置为10002。
spring: - Spring框架相关的配置开始。
main: - Spring的主配置，用于配置Spring应用程序的主要属性。
allow-circular-references: true - 允许循环引用。当Bean之间存在循环依赖时，设置为true可以解决循环依赖问题。
application: - 应用程序相关的配置开始。
name: dispatcher - 应用程序的名称，这里配置为dispatcher。
mqtt: - MQTT相关的配置开始。
hostUrl: tcp://ip:1883 - MQTT服务器的主机URL，表示MQTT消息将通过TCP协议发送到指定的IP地址和端口。这里配置为tcp://ip:1883，表示连接到运行在IP地址上的MQTT服务器，端口号为1883。
username: dev - MQTT客户端的用户名，用于身份验证。这里配置为dev。
password: dev - MQTT客户端的密码，用于身份验证。这里配置为dev。
client-id: MQTT-CLIENT-DEV - MQTT客户端的唯一标识符。这里配置为MQTT-CLIENT-DEV。
cleanSession: true - MQTT会话的清理标志。设置为true表示在连接关闭时清除会话中的所有消息。
reconnect: true - MQTT客户端的重连标志。设置为true表示在连接断开时自动尝试重新连接MQTT服务器。
timeout: 100 - MQTT客户端的超时时间，单位为毫秒。这里配置为100毫秒。
keepAlive: 100 - MQTT客户端的心跳保持时间，单位为毫秒。这里配置为100毫秒。
defaultTopic: server/dev/report - MQTT客户端的默认主题，用于接收消息。这里配置为server/dev/report。
serverTopic: server/dev/report - 服务器端发布消息的主题。这里配置为server/dev/report。
isOpen: true - 是否开启MQTT功能，这里配置为true表示开启MQTT功能。
qos: 0 - MQTT消息的QoS（服务质量）等级，这里配置为0表示最多一次的传输保障。

回调实现

package com.adn.callback.Impl;import com.adn.client.MqttAcceptClient;
import com.adn.common.config.MqttProperties;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.io.UnsupportedEncodingException;/*** @Description : MQTT接受服务的回调类* @Author : adn*/
@Component
public class MqttAcceptCallback implements MqttCallbackExtended {private static final Logger logger = LoggerFactory.getLogger(MqttAcceptCallback.class);@Autowiredprivate MqttAcceptClient mqttAcceptClient;@Autowiredprivate MqttProperties mqttProperties;/*** 客户端断开后触发** @param throwable*/@Overridepublic void connectionLost(Throwable throwable) {logger.info("连接断开，可以重连");if (MqttAcceptClient.client == null || !MqttAcceptClient.client.isConnected()) {logger.info("【emqx重新连接】....................................................");mqttAcceptClient.reconnection();}}/*** 客户端收到消息触发** @param topic       主题* @param mqttMessage 消息*/@Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {logger.info("【接收消息主题】:" + topic);logger.info("【接收消息Qos】:" + mqttMessage.getQos());logger.info("【接收消息内容】:" + new String(mqttMessage.getPayload()));//        int i = 1/0;}/*** 发布消息成功** @param token token*/@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {String[] topics = token.getTopics();for (String topic : topics) {logger.info("向主题【" + topic + "】发送消息成功！");}try {MqttMessage message = token.getMessage();byte[] payload = message.getPayload();String s = new String(payload, "UTF-8");logger.info("【消息内容】:" + s);} catch (Exception e) {logger.error("MqttAcceptCallback deliveryComplete error,message:{}", e.getMessage());e.printStackTrace();}}/*** 连接emq服务器后触*/@Overridepublic void connectComplete(boolean b, String s) {logger.info("============================= 客户端【" + MqttAcceptClient.client.getClientId() + "】连接成功！=============================");// 以/#结尾表示订阅所有以test开头的主题// 订阅所有机构主题mqttAcceptClient.subscribe(mqttProperties.getDefaultTopic(), 0);}
}

客户端

package com.adn.client;import com.adn.callback.Impl.MqttAcceptCallback;
import com.adn.common.config.MqttProperties;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** @Description : MQTT接受服务的客户端* @Author : adn*/
@Component
public class MqttAcceptClient {private static final Logger logger = LoggerFactory.getLogger(MqttAcceptClient.class);@Autowiredprivate MqttAcceptCallback mqttAcceptCallback;@Autowiredprivate MqttProperties mqttProperties;public static MqttClient    client;private static MqttClient getClient() {return client;}private static void setClient(MqttClient client) {MqttAcceptClient.client = client;}/*** 客户端连接*/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.setClient(client);// 设置回调client.setCallback(mqttAcceptCallback);client.connect(options);} catch (Exception e) {logger.error("MqttAcceptClient connect error,message:{}", e.getMessage());e.printStackTrace();}}/*** 重新连接*/public void reconnection() {try {client.connect();} catch (MqttException e) {logger.error("MqttAcceptClient reconnection error,message:{}", e.getMessage());e.printStackTrace();}}/*** 订阅某个主题** @param topic 主题* @param qos   连接方式*/public void subscribe(String topic, int qos) {logger.info("========================【开始订阅主题:" + topic + "】========================");try {client.subscribe(topic, qos);} catch (MqttException e) {logger.error("MqttAcceptClient subscribe error,message:{}", e.getMessage());e.printStackTrace();}}/*** 取消订阅某个主题** @param topic*/public void unsubscribe(String topic) {logger.info("========================【取消订阅主题:" + topic + "】========================");try {client.unsubscribe(topic);} catch (MqttException e) {logger.error("MqttAcceptClient unsubscribe error,message:{}", e.getMessage());e.printStackTrace();}}
}

使用配置

MqttCondition

package com.adn.common.config.Impl;import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.annotation.Condition;
import org.springframework.context.annotation.ConditionContext;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotatedTypeMetadata;/*** @Description : 自定义配置,通过这个配置，来控制启动项目的时候是否启动mqtt* @Author : adn*/
public class MqttCondition implements Condition {@Overridepublic boolean matches(ConditionContext context, AnnotatedTypeMetadata annotatedTypeMetadata) {//1、能获取到ioc使用的beanfactoryConfigurableListableBeanFactory beanFactory = context.getBeanFactory();//2、获取类加载器ClassLoader classLoader = context.getClassLoader();//3、获取当前环境信息Environment environment = context.getEnvironment();String isOpen = environment.getProperty("mqtt.isOpen");return Boolean.valueOf(isOpen);}
}

MqttConfig

package com.adn.common.config;import com.adn.client.MqttAcceptClient;
import com.adn.common.config.Impl.MqttCondition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;/*** @Description : 启动服务的时候开启监听客户端* @Author : adn*/
@Configuration
public class MqttConfig {@Autowiredprivate MqttAcceptClient mqttAcceptClient;/*** 订阅mqtt** @return*/@Conditional(MqttCondition.class)@Beanpublic MqttAcceptClient getMqttPushClient() {mqttAcceptClient.connect();return mqttAcceptClient;}
}

MqttProperties

package com.adn.common.config;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;/***  MQTT配置信息* @Author : adn*/
@Component
@ConfigurationProperties("mqtt")
@Data
public class MqttProperties {/*** 用户名*/private String  username;/*** 密码*/private String  password;/*** 连接地址*/private String  hostUrl;/*** 客户端Id，同一台服务器下，不允许出现重复的客户端id*/private String  clientId;/*** 默认连接主题，以/#结尾表示订阅所有以test开头的主题*/private String  defaultTopic;/*** 默认服务器发送主题前缀，格式：server:${env}:report:${topic}*/private String  serverTopic;/*** 超时时间*/private int     timeout;/*** 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端* 发送个消息判断客户端是否在线，但这个方法并没有重连的机制*/private int     keepAlive;/*** 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连* 接记录，这里设置为true表示每次连接到服务器都以新的身份连接*/private Boolean cleanSession;/*** 是否断线重连*/private Boolean reconnect;/*** 启动的时候是否关闭mqtt*/private Boolean isOpen;/*** 连接方式*/private Integer qos;/*** 获取默认主题，以/#结尾表示订阅所有以test开头的主题* @return*/public String getDefaultTopic() {return defaultTopic + "/#";}/*** 获取服务器发送主题，格式：server/${env}/report/${topic}* @param topic* @return*/public String getServerTopic(String topic) {return serverTopic + "/" + topic;}
}

启动测试

查看可视化面板，成功创建连接。

如果需要更换订阅的主题可以在配置文件中更换以及在建立连接的时候更换。

发布者代码demo展示

发布者的配置pom和application.yml文件中内容与上面的订阅者一样。参考上面即可

代码结构展示：

回调类

package com.adn.callback.Impl;import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;/*** @Description : MQTT发送客户端的回调类* @Author : adn*/
@Component
public class MqttSendCallBack implements MqttCallbackExtended {private static final Logger logger = LoggerFactory.getLogger(MqttSendCallBack.class);/*** 客户端断开后触发** @param throwable*/@Overridepublic void connectionLost(Throwable throwable) {logger.info("连接断开，可以重连");}/*** 客户端收到消息触发** @param topic       主题* @param mqttMessage 消息*/@Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {logger.info("【接收消息主题】: " + topic);logger.info("【接收消息Qos】: " + mqttMessage.getQos());logger.info("【接收消息内容】: " + new String(mqttMessage.getPayload()));}/*** 发布消息成功** @param token token*/@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {String[] topics = token.getTopics();for (String topic : topics) {logger.info("向主题【" + topic + "】发送消息成功！");}try {MqttMessage message = token.getMessage();byte[] payload = message.getPayload();String s = new String(payload, "UTF-8");logger.info("【消息内容】:" + s);} catch (Exception e) {logger.error("MqttSendCallBack deliveryComplete error,message:{}", e.getMessage());e.printStackTrace();}}/**** 连接emq服务器后触发** @param b* @param s*/@Overridepublic void connectComplete(boolean b, String s) {
//        logger.info("============================= 客户端【" + MqttAcceptClient.client.getClientId() + "】连接成功！=============================");}}

发送客户端

package com.adn.client;import com.adn.callback.Impl.MqttSendCallBack;
import com.adn.common.config.MqttProperties;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.UUID;/*** @Description : MQTT发送客户端* @Author : adn*/
@Component
public class MqttSendClient {private static final Logger logger       = LoggerFactory.getLogger(MqttSendClient.class);@Autowiredprivate MqttSendCallBack mqttSendCallBack;@Autowiredprivate MqttProperties mqttProperties;public MqttClient connect() {MqttClient client = null;try {String uuid = UUID.randomUUID().toString().replaceAll("-", "");client = new MqttClient(mqttProperties.getHostUrl(), uuid, 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);// 设置回调client.setCallback(mqttSendCallBack);client.connect(options);} catch (Exception e) {logger.error("MqttSendClient connect error,message:{}", e.getMessage());e.printStackTrace();}return client;}/*** 发布消息** @param retained 是否保留* @param topic 主题，格式： server:${env}:report:${topic}* @param content 消息内容*/public void publish(boolean retained, String topic, String content) {MqttMessage message = new MqttMessage();message.setQos(mqttProperties.getQos());message.setRetained(retained);message.setPayload(content.getBytes());MqttDeliveryToken token;MqttClient mqttClient = connect();try {mqttClient.publish(mqttProperties.getServerTopic(topic), message);} catch (MqttException e) {logger.error("MqttSendClient publish error,message:{}", e.getMessage());e.printStackTrace();} finally {disconnect(mqttClient);close(mqttClient);}}/*** 关闭连接** @param mqttClient*/public static void disconnect(MqttClient mqttClient) {try {if (mqttClient != null){mqttClient.disconnect();}} catch (MqttException e) {logger.error("MqttSendClient disconnect error,message:{}", e.getMessage());e.printStackTrace();}}/*** 释放资源** @param mqttClient*/public static void close(MqttClient mqttClient) {try {if (mqttClient != null) {mqttClient.close();}} catch (MqttException e) {logger.error("MqttSendClient close error,message:{}", e.getMessage());e.printStackTrace();}}
}

配置文件以及配置类

MqttProperties

package com.adn.common.config;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;/***  MQTT配置信息* @Author : adn*/
@Component
@ConfigurationProperties("mqtt")
@Data
public class MqttProperties {/*** 用户名*/private String  username;/*** 密码*/private String  password;/*** 连接地址*/private String  hostUrl;/*** 客户端Id，同一台服务器下，不允许出现重复的客户端id*/private String  clientId;/*** 默认连接主题，以/#结尾表示订阅所有以test开头的主题*/private String  defaultTopic;/*** 默认服务器发送主题前缀，格式：server:${env}:report:${topic}*/private String  serverTopic;/*** 超时时间*/private int     timeout;/*** 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端* 发送个消息判断客户端是否在线，但这个方法并没有重连的机制*/private int     keepAlive;/*** 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连* 接记录，这里设置为true表示每次连接到服务器都以新的身份连接*/private Boolean cleanSession;/*** 是否断线重连*/private Boolean reconnect;/*** 启动的时候是否关闭mqtt*/private Boolean isOpen;/*** 连接方式*/private Integer qos;/*** 获取默认主题，以/#结尾表示订阅所有以test开头的主题* @return*/public String getDefaultTopic() {return defaultTopic + "/#";}/*** 获取服务器发送主题，格式：server/${env}/report/${topic}* @param topic* @return*/public String getServerTopic(String topic) {return serverTopic + "/" + topic;}
}

MqttCondition

package com.adn.common.config.Impl;import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.annotation.Condition;
import org.springframework.context.annotation.ConditionContext;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotatedTypeMetadata;/*** @Description : 自定义配置,通过这个配置，来控制启动项目的时候是否启动mqtt* @Author : adn* 0 16:32*/
public class MqttCondition implements Condition {@Overridepublic boolean matches(ConditionContext context, AnnotatedTypeMetadata annotatedTypeMetadata) {//1、能获取到ioc使用的beanfactoryConfigurableListableBeanFactory beanFactory = context.getBeanFactory();//2、获取类加载器ClassLoader classLoader = context.getClassLoader();//3、获取当前环境信息Environment environment = context.getEnvironment();String isOpen = environment.getProperty("mqtt.isOpen");return Boolean.valueOf(isOpen);}
}

测试的controller

package com.adn.controller;import com.adn.client.MqttSendClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;/*** @Description : 测试类* @Author : adn*/
@RestController
@RequestMapping("/mqtt")
public class MqttController {@Autowiredprivate MqttSendClient mqttSendClient;@GetMapping(value = "/publishTopic")public String publishTopic(String topic, String sendMessage) {topic = "client/dev/report/test";System.out.println("topic:" + topic);System.out.println("message:" + sendMessage);for (int i = 0; i < 10000; i++) {this.mqttSendClient.publish(false, topic, sendMessage);}return "topic:" + topic + "message:" + sendMessage;}}