一、介绍
1、MQTT
MQTT(Message Queuing Telemetry Transport, 消息队列遥测传输协议),是一种基于发布/订阅(publish/subscribe)模式的"轻量级"通讯协议,该协议构建于TCP/IP协议上,由IBM在1999年发布。MQTT最大优点在于,可以以极少的代码和有限的带宽,为远程连接设备提过实时可靠的消息服务,作为一种低开销、低带宽占用的即时通讯协议,使其在物联网、小型设备、移动应用等方面有较广泛的应用。
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的,这些特点使它适用范围非常广泛。在很多情况下,包括受限的环境中,如:机器与机器(M2M)通信和物联网(loT)。其在,通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
特点:
使用发布/订阅消息模式,提供一对多的消息发布,解除应用程序耦合;
对负载内容屏蔽的消息传输;
使用 TCP/IP 提供网络连接;
有三种消息发布服务质量:
小型传输,开销很小(固定长度的头部是 2 字节),协议交换最小化,以降低网络流量;
使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制。
2、EMQX
EMQX 是一个「无限连接,任意集成,随处运行」大规模分布式物联网接入平台。
EMQX 企业版提供一体化的分布式 MQTT 消息服务和强大的 IoT 规则引擎,为高可靠、高性能的物联网实时数据移动、处理和集成提供动力,助力企业快速构建关键业务的 IoT 平台与应用。附下载地址: https://www.emqx.com/zh/try?product=enterprise 可以自行下载对应版本运行
优势:
海量连接:单节点支持 500 万 MQTT 设备连接,集群可水平扩展至支持 1 亿并发的 MQTT 连接。
高可靠:弹性伸缩,无单点故障。内置 RocksDB 可靠地持久化 MQTT 消息,确保无数据损失。
数据安全:端到端数据加密(支持国密),细粒度访问控制,保障数据安全,满足企业合规需求。
多协议:支持 MQTT、HTTP、QUIC、WebSocket、LwM2M/CoAP 或专有协议连接任何设备。
高性能:单节点支持每秒实时接收、处理与分发数百万条的 MQTT 消息。毫秒级消息交付时延。
易运维:图形化配置、操作与管理,实时监测运行状态。支持 MQTT 跟踪进行端到端问题分析。
3、Mria 集群架构
支持全新的 Mria 集群架构,在此架构下 EMQX 水平扩展性得到指数级提升,单个集群可以轻松支持 1 亿 MQTT 连接,这使得 EMQX 5.0 成为目前全球最具扩展性的 MQTT Broker。
在构建满足用户业务需求的更大规模集群的同时,Mria 架构还能够降低大规模部署下的脑裂风险以及脑裂后的影响,以提供更加稳定可靠的物联网数据接入服务。
具体可以查看官方文档: https://docs.emqx.com/zh/enterprise/v5.1/deploy/cluster/create-cluster.html
4、MQTTX
MQTTX 是由 EMQ 开发的一款开源跨平台 MQTT 5.0 桌面客户端,它兼容 macOS,Linux 以及 Windows 系统。MQTTX 的用户界面 UI 采用聊天式设计,使得操作逻辑更加简明直观。它支持用户快速创建和保存多个 MQTT 连接,便于测试 MQTT/MQTTS 连接,以及 MQTT 消息的订阅和发布。
主要功能
采用聊天界面设计,使得操作更加简单明了
跨平台兼容,支持在 Windows,macOS,Linux 系统上运行
100% 兼容 MQTT v5.0,v3.1.1 和 v3.1 协议
订阅的 MQTT 主题支持自定义颜色标签
支持单向和双向 SSL 认证,同时支持 CA 和自签名证书
支持通过 WebSocket 连接 MQTT 服务器
支持 Hex, Base64, JSON, Plaintext 等 Payload 格式转换
自定义脚本支持模拟 MQTT 发布/订阅测试
提供完整的日志记录功能
多语言支持:简体中文、英语、日语、土耳其语及匈牙利语 ??? ??? ??? ??? ???
自由切换 Light、Dark、Night 三种主题模式
二、实战
1、引入maven依赖:
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.2</version>
</dependency>
# EMQX配置
emqx:
# EMQX服务地址,端口号默认18083
url: http://127.0.0.1:18083
# 认证用户名
username: admin
# 密码
password: admin123456
/**
* EMQX 登录账号密码配置
*/
@Data
@Configuration
@ConfigurationProperties(prefix = "emqx")
public class EmqxConfig {
private String url;
private String username;
private String password;
}
spring:
# MQTT配置
mqtt:
# MQTT服务地址,端口号默认1883,如果有多个,用逗号隔开
host-url: tcp://127.0.0.1:1883
# 用户名
username: admin
# 密码
password: admin123456
# 客户端id(不能重复)
client-id: real-mqtt-client
# MQTT默认的消息推送主题,实际可在调用接口时指定
default-topic: topic
@Configuration
@ConfigurationProperties(prefix = "spring.mqtt")
@Data
public class MqttConfig {
private String username;
private String password;
private String hostUrl;
private String clientId;
private String defaultTopic;
}
MQTT客户端连接工厂类文章来源:https://uudwc.com/A/edR9m
@Slf4j
@Component
public class MqttFactory {
public static ConcurrentHashMap<String, MqttClient> clientMap = new ConcurrentHashMap<>();
@Autowired
private MqttConfig mqttConfig;
@Autowired
private RealPersonAccessDeviceMapper realPersonAccessDeviceMapper;
/**
* 在bean初始化后连接到服务器
*/
@PostConstruct
public void init() {
String mqttStartFlag = ParamResolver.getStr(RealCommonConstants.MQTT_START_FLAG);
if (StrUtil.equals(mqttStartFlag, CommonConstants.SYS_YES_NO_Y)) {
// 初始化订阅主题
initSubscribeTopic(getInstance());
}
}
/**
* 初始化客户端
*/
public MqttClient getInstance() {
MqttClient client = null;
if (clientMap.get(mqttConfig.getClientId()) == null) {
try {
client = new MqttClient(mqttConfig.getHostUrl(), mqttConfig.getClientId());
// MQTT配置对象
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
// 设置自动重连, 其它具体参数可以查看MqttConnectOptions
mqttConnectOptions.setAutomaticReconnect(true);
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录,这里设置为true表示每次连接到服务器都以新的身份连接
// mqttConnectOptions.setCleanSession(true);
// 设置超时时间 单位为秒
mqttConnectOptions.setConnectionTimeout(10);
mqttConnectOptions.setUserName(mqttConfig.getUsername());
mqttConnectOptions.setPassword(mqttConfig.getPassword().toCharArray());
// mqttConnectOptions.setServerURIs(new String[]{url});
// 设置会话心跳时间 单位为秒
mqttConnectOptions.setKeepAliveInterval(10);
// 设置“遗嘱”消息的话题,若客户端与服务器之间的连接意外中断,服务器将发布客户端的“遗嘱”消息。
// mqttConnectOptions.setWill("willTopic", "offline".getBytes(), 2, false);
if (!client.isConnected()) {
client.connect(mqttConnectOptions);
}
client.setCallback(new MqttCallBack());
log.info("MQTT创建client成功={}", JSONObject.toJSONString(client));
clientMap.put(mqttConfig.getClientId(), client);
} catch (MqttException e) {
log.error("MQTT连接消息服务器[{}]失败", mqttConfig.getClientId() + "-" + mqttConfig.getHostUrl());
}
} else {
client = clientMap.get(mqttConfig.getClientId());
log.info("MQTT从map里获取到client,clientId=" + mqttConfig.getClientId());
// TODO 已采用自动重连策略
// log.info("MQTT从map里获取到client={}", JSONObject.toJSONString(client));
// if (!client.isConnected()) {
// initSubscribeTopic(client);
// 如果缓存里的client已经断开,则清除该缓存,再重新创建客户端连接
// clientMap.remove(mqttConfig.getClientId());
// this.getInstance();
// }
}
return client;
}
/**
* 初始化订阅主题
* <p>
* 消息等级,和主题数组一一对应,服务端将按照指定等级给订阅了主题的客户端推送消息
*/
public void initSubscribeTopic(MqttClient client) {
// 查询所有宇泛mqtt设备并订阅主题
List<RealPersonAccessDevice> deviceList = realPersonAccessDeviceMapper.selectList(Wrappers.<RealPersonAccessDevice>lambdaQuery()
.eq(RealPersonAccessDevice::getDeviceProducerType, RealCommonConstants.DeviceProducerType.UNIUBI.getValue())
.and(wrapper -> wrapper
.eq(RealPersonAccessDevice::getProtocolType, RealCommonConstants.ProtocolType.MQTT.getValue())
.or()
.eq(RealPersonAccessDevice::getProtocolType, RealCommonConstants.ProtocolType.UMQTT.getValue()))
.eq(RealPersonAccessDevice::getStatus, CommonConstants.STATUS_ENABLE));
if (CollectionUtil.isNotEmpty(deviceList)) {
// 订阅设备发布消息主题
List<String> upstreamTopics = new ArrayList<>();
List<Integer> upstreamQos = new ArrayList<>();
for (RealPersonAccessDevice device : deviceList) {
if (StrUtil.equals(device.getProtocolType(), RealCommonConstants.ProtocolType.MQTT.getValue())) {
upstreamTopics.add(String.format(MqttCommonConstants.DEFAULT, device.getDeviceNo()));
upstreamQos.add(1);
upstreamTopics.add(String.format(MqttCommonConstants.UPSTREAM, device.getDeviceNo()));
upstreamQos.add(0);
upstreamTopics.add(String.format(MqttCommonConstants.WILL, device.getDeviceNo()));
upstreamQos.add(1);
} else if (StrUtil.equals(device.getProtocolType(), RealCommonConstants.ProtocolType.UMQTT.getValue())) {
upstreamTopics.add(String.format(UMqttCommonConstants.EVENT, device.getDeviceNo()));
upstreamQos.add(1);
upstreamTopics.add(String.format(UMqttCommonConstants.ONLINE));
upstreamQos.add(1);
upstreamTopics.add(String.format(UMqttCommonConstants.RESPONSE, device.getDeviceNo()));
upstreamQos.add(2);
upstreamTopics.add(String.format(UMqttCommonConstants.UPLOAD, device.getDeviceNo()));
upstreamQos.add(0);
}
}
try {
client.subscribe(upstreamTopics.toArray(new String[upstreamTopics.size()]), upstreamQos.stream().mapToInt(Integer::intValue).toArray());
} catch (MqttException e) {
e.printStackTrace();
}
}
}
}
业务方法文章来源地址https://uudwc.com/A/edR9m
@Slf4j
@Data
@Configuration
public class UMqttClientService {
private final MqttFactory mqttFactory;
private final StringRedisTemplate redisTemplate;
/**
* 订阅主题
*/
public void subscribeTopic(String deviceNo) {
try {
// 订阅设备发布消息主题
List<String> upstreamTopics = new ArrayList<>();
upstreamTopics.add(String.format(UMqttCommonConstants.EVENT, deviceNo));
upstreamTopics.add(UMqttCommonConstants.ONLINE);
upstreamTopics.add(String.format(UMqttCommonConstants.RESPONSE, deviceNo));
upstreamTopics.add(String.format(UMqttCommonConstants.UPLOAD, deviceNo));
int[] upstreamQos = {1, 1, 2, 0};
mqttFactory.getInstance().subscribe(upstreamTopics.toArray(new String[0]), upstreamQos);
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 取消订阅主题
*/
public void stopSubscribeTopic(String deviceNo) {
try {
// 取消订阅设备发布消息主题
List<String> upstreamTopics = new ArrayList<>();
upstreamTopics.add(String.format(UMqttCommonConstants.EVENT, deviceNo));
upstreamTopics.add(UMqttCommonConstants.ONLINE);
upstreamTopics.add(String.format(UMqttCommonConstants.RESPONSE, deviceNo));
upstreamTopics.add(String.format(UMqttCommonConstants.UPLOAD, deviceNo));
mqttFactory.getInstance().unsubscribe(upstreamTopics.toArray(new String[0]));
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 断开连接
*/
public void disConnect() {
try {
mqttFactory.getInstance().disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 订阅主题
*/
public void subscribe(String topic, int qos) {
try {
mqttFactory.getInstance().subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 发布请求设备消息
*
* @param deviceNo 设备编号
* @param message 消息
*/
public void publish(String deviceNo, String message) {
publish(1, false, String.format(UMqttCommonConstants.REQUEST, deviceNo), message);
}
/**
* 发布请求设备消息
*/
public void publish(UMqttPublishDate publishDate) {
// 将消息id和方法名存到redis中:缓存3分钟
redisTemplate.opsForValue().set(UMqttCommonConstants.UMQTT_MSG_REDIS_KEY + publishDate.getId(),
JSON.toJSONString(publishDate), 24, TimeUnit.HOURS);
publish(1, false, String.format(UMqttCommonConstants.REQUEST, publishDate.getDeviceNo()), publishDate.getMessage());
}
/**
* 发布响应设备消息
*/
public void publishResponse(UMqttPublishDate publishDate) {
// 将消息id和方法名存到redis中:缓存3分钟
redisTemplate.opsForValue().set(UMqttCommonConstants.UMQTT_MSG_REDIS_KEY + publishDate.getId(),
JSON.toJSONString(publishDate), 24, TimeUnit.HOURS);
publish(1, false, String.format(UMqttCommonConstants.RESPONSE, publishDate.getDeviceNo()), publishDate.getMessage());
}
/**
* 发布消息
*
* @param qos qos
* @param retained retained
* @param topic 主题
* @param message 消息
*/
public void publish(int qos, boolean retained, String topic, String message) {
log.info("发布消息topic:" + topic);
log.info("发布消息message:" + message);
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setQos(qos);
mqttMessage.setRetained(retained);
mqttMessage.setPayload(message.getBytes());
//主题的目的地,用于发布/订阅信息
MqttTopic mqttTopic = mqttFactory.getInstance().getTopic(topic);
//提供一种机制来跟踪消息的传递进度
//用于在以非阻塞方式(在后台运行)执行发布是跟踪消息的传递进度
MqttDeliveryToken token;
try {
//将指定消息发布到主题,但不等待消息传递完成,返回的token可用于跟踪消息的传递状态
//一旦此方法干净地返回,消息就已被客户端接受发布,当连接可用,将在后台完成消息传递。
token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
} catch (MqttException e) {
e.printStackTrace();
}
}
}