Java中间件接入之MQTT组件

1.MQTT介绍

MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅消息传输协议，专门用于物联网（IoT）领域的通信。它旨在提供可靠且高效的消息传递机制，使设备和应用程序能够在低带宽、不稳定网络环境下进行实时通信。

MQTT协议采用了发布/订阅模式，其中包含两个主要角色：发布者（Publisher）和订阅者（Subscriber）。发布者负责将消息发布到特定的主题（Topic），而订阅者则通过订阅特定主题来接收相关消息。这种松耦合的通信方式使得系统能够轻松地扩展和管理。

MQTT协议具有以下特点：

• 轻量级：MQTT协议使用简单的二进制协议头，消息开销很小，适合在低带宽、低功耗设备上使用。
• 可靠性：MQTT支持确认和重传机制，确保消息在不稳定网络中的可靠传输。
• 异步通信：发布者和订阅者不需要互相了解对方的存在，使得系统更加松耦合。
• 灵活的消息过滤：订阅者可以使用通配符来过滤感兴趣的主题，灵活地选择接收哪些消息。
• 安全性：MQTT支持通过TLS/SSL协议进行加密传输，并且可以使用认证机制进行身份验证。

由于其轻量级和简单易用的特点，MQTT被广泛用于物联网场景中，如传感器数据采集、远程监控、智能家居等。

官方资料链接：

MQTT ORG
EMQX

2.搭建MQTT服务器

1.Windows

1. 下载emqx-5.1.0-windows-amd64.zip，解压
2. 命令行下进入解压路径，执行以下命令启动 emqx

./emqx/bin/emqx start

2.Ubuntu

1.通过以下命令配置 EMQX Apt 源：

curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash

2.运行以下命令安装 EMQX：

sudo apt-get install emqx

3.运行以下命令启动 EMQX：

sudo systemctl start emqx

3.Docker

1.运行以下命令获取 Docker 镜像：

docker pull emqx/emqx:5.4.0

2.运行以下命令启动 Docker 容器。

docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.4.0

4.其他方式

参考：EMQX文档

3.mqtt可视化客户端MQTTX及快速使用教程

MQTTX下载地址：MQTT 5.0 客户端工具

快速使用教程参考文章：Windows版mqtt可视化客户端快速使用教程

4.SpringBoot接入MQTT

1、maven依赖

		<!--MQTT依赖库-->
		<dependency>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-core</artifactId>
			<version>5.4.2</version>
		</dependency>
		<dependency>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-mqtt</artifactId>
			<version>5.5.5</version>
		</dependency>

2、MQTT配置

在bootstrap.yml中增加MQTT配置信息，可根据自身情况修改相关字段

#是否启用MQTT服务
spring.mqtt.enable=true
#MQTT-服务端地址（本地服务器）
spring.mqtt.url=tcp://127.0.0.1:1883
#MQTT-服务端用户名
spring.mqtt.username=admin
#MQTT-服务端密码（本地服务器）
spring.mqtt.password=public
#MQTT-客户端clientid
spring.mqtt.clientid=test001
#MQTT-Warning默认主题
spring.mqtt.topicWarning=topic_warning

3、MQTT组件具体代码

1.定义通道名字

public class ChannelName {
   

    /**
     * 订阅的bean名称
     */
    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";
    /**
     * 发布的bean名称
     */
    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
}

2.消息发布器

import com.funfan.autoCodeDemo.component.mqtt.model.ChannelName;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import