Vert.x框架实现MQTT服务，代码实现及其详解

2024-02-07 08:12
文章标签 代码 实现 服务 详解 框架 mqtt vert

本文主要是介绍Vert.x框架实现MQTT服务，代码实现及其详解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

简单的总结

先做一个简单的近日总结吧
最近在学习怎么用vertx框架实现一个mqtt服务和物联网平台进一步的搭建
这两天又折腾了笔记软件，想从语雀迁到obsidian，也花了不少时间，想着最近都没怎么写过博客，就写一写自己这几天学习vertx框架实现mqtt服务的收获和学习到东西

Vert.x MQTT

MQTT是基于发布/订阅(publish/subscribe)模式的”轻量级“通信协议，该协议构建于TCP/IP协议之上，由于其低开销，低带宽占用的特点，非常适合在物联网设备通信方面使用。
关于MQTT相关的基础知识我不再赘述，如果还不清楚的同学可以看看这个文档，快速了解一下相关知识。MQTT 协议入门：基础知识和快速教程

而Vert.x MQTT是一个使用Vert.x实现MQTT协议中客户端与服务端的组件，里边有一些基础的api，可以利用api写出自己想要的服务。以下是其官方文档，可以参考一下，虽然是汉化过的文档，但是我看的过程还是比较困难，可能是作者太菜了。Vert.x MQTT

这里我只用了Vertx MQTT的服务端组件实现了MQTT服务，客户端部分采用了MQTTX创建虚拟设备。MQTTX是一款客户端工具，可以用于创建虚拟设备快速进行测试，MQTT 客户端工具演示 | EMQX 文档

代码实现

先分方法放出来，最后再放完整代码

start()方法

代码的入口，创建了一个MQTT服务器并进行了监听。在接收到来自客户端的连接请求时，会调用不同的处理方法来处理订阅、退订、收到消息和断开连接等操作。

  @Overridepublic void start() throws Exception {MqttServer mqttServer = MqttServer.create(vertx);topicSubscribers = new HashMap<>();subscriptions=new HashMap<>();mqttServer.endpointHandler(endpoint -> {//使用lambda表达式实现了endpointHandler方法，传入参数endpoint；System.out.println("MQTT client [" + endpoint.clientIdentifier() + "] request to connect, clean session = " + endpoint.isCleanSession());if (endpoint.auth() != null) {System.out.println("[username = " + endpoint.auth().getUsername() + ", password = " + endpoint.auth().getPassword() + "]");}System.out.println("[properties = " + endpoint.connectProperties() + "]");// accept connection from the remote clientendpoint.accept(false);SubscribeHandle(endpoint);UnSubscribeHandle(endpoint);ReceiveHandle(endpoint);DisConnectHandle(endpoint);});mqttServer.listen(ar -> {if (ar.succeeded()) {System.out.println("MQTT server is listening on port " + ar.result().actualPort());} else {System.out.println("Error on starting the server");ar.cause().printStackTrace();}});}

DisConnectHandle方法

用于处理设备断开连接的情况。当设备发送断开连接消息时，会从订阅关系列表中删除该设备，并从订阅列表中删除该设备的所有订阅。

  private void DisConnectHandle(MqttEndpoint endpoint) {endpoint.disconnectMessageHandler(disconnectMessage -> {System.out.println("Received disconnect from client, reason code = " + disconnectMessage.code());System.out.println("Client "+endpoint.auth().getUsername()+"disconnect with");//对两个映射订阅关系的列表进行更新for (String topic:subscriptions.get(endpoint)){topicSubscribers.get(topic).remove(endpoint);System.out.println("["+topic+"]");}subscriptions.remove(endpoint);});}

SubscribeHandle方法

用于处理订阅消息。当设备发送订阅消息时，会检查订阅的主题是否合法。如果主题合法，则将设备添加到订阅关系列表和订阅列表中，并向设备发送订阅确认消息。

    private void SubscribeHandle(MqttEndpoint endpoint) {endpoint.subscribeHandler(subscribe -> {Boolean IsValidTopic=false;//存储订阅消息中要订阅的topic的列表List<MqttTopicSubscription> topicSubscriptions = subscribe.topicSubscriptions();//存储订阅topic Qos级别的列表List<MqttSubAckReasonCode> reasonCodes = new ArrayList<>();//遍历列表for (MqttTopicSubscription s : topicSubscriptions) {//topicString topic = s.topicName();//Qos级别MqttQoS qos = s.qualityOfService();//判断topic是否合法if (!isValidTopic(topic)){//不合法则向设备发送消息endpoint.publish(topic, Buffer.buffer("非法topic，topic不可包含空格"), qos, false, false);continue;}else {IsValidTopic=true;}System.out.println("Subscription for " + topic + " with QoS " + qos);reasonCodes.add(MqttSubAckReasonCode.qosGranted(qos));//判断是否已有此topic，如果有则直接添加，没有则新建键值对if (!topicSubscribers.containsKey(topic)) {topicSubscribers.put(topic, new ArrayList<MqttEndpoint>());}topicSubscribers.get(topic).add(endpoint);//同上if (!subscriptions.containsKey(endpoint)) {subscriptions.put(endpoint, new ArrayList<String>());}subscriptions.get(endpoint).add(topic);}if(IsValidTopic){endpoint.subscribeAcknowledge(subscribe.messageId(), reasonCodes, MqttProperties.NO_PROPERTIES);}});}

UnSubscribeHandle方法

用于处理退订消息。当设备发送退订消息时，会从订阅关系列表中删除该设备，并检查是否需要删除订阅列表中的主题。

private void UnSubscribeHandle(MqttEndpoint endpoint) {endpoint.unsubscribeHandler(unsubscribe -> {//遍历要退订的topicfor (String unsubscribedTopic : unsubscribe.topics()) {topicSubscribers.get(unsubscribedTopic).remove(endpoint);//如果某topic的订阅列表为空，删除topicif (topicSubscribers.get(unsubscribedTopic).size() == 0) {topicSubscribers.remove(unsubscribedTopic);}subscriptions.get(endpoint).remove(unsubscribedTopic);//同上if (subscriptions.get(endpoint).size()==0){subscriptions.remove(endpoint);}System.out.println("unsubscribed :" + endpoint.auth().getUsername() + "for" + unsubscribedTopic);}endpoint.unsubscribeAcknowledge(unsubscribe.messageId());});}

ReceiveHandle方法

用于处理收到的消息。当设备发布消息时，会检查发布的主题是否合法。如果主题合法，则遍历订阅关系列表，将消息发布给订阅了匹配主题的设备。

private void ReceiveHandle(MqttEndpoint endpoint) {endpoint.publishHandler(publish -> {String topic = publish.topicName();Buffer payload = publish.payload();//对topic的合法性进行判断if (!isValidTopic(topic)){endpoint.publish(topic, Buffer.buffer("非法topic，topic不可包含空格"), MqttQoS.AT_MOST_ONCE, false, false);return;}//记录日志接收到设备发布的消息System.out.println("Received message [" + publish.payload().toString(Charset.defaultCharset()) + "] with QoS [" + publish.qosLevel() + "]");if (publish.qosLevel() == MqttQoS.AT_LEAST_ONCE) {endpoint.publishAcknowledge(publish.messageId());} else if (publish.qosLevel() == MqttQoS.EXACTLY_ONCE) {endpoint.publishReceived(publish.messageId());}//遍历订阅关系，进行消息发布for (Map.Entry<String, List<MqttEndpoint>> entry : topicSubscribers.entrySet()) {String subscribedTopic = entry.getKey();//被订阅的topicList<MqttEndpoint> subscribers = entry.getValue();//订阅上方topic的订阅者//判断消息发布的topic是否能和被设备订阅的topic按照规则匹配if (isTopicMatch(subscribedTopic, topic)) {//若匹配，则遍历topic订阅者列表，并进行消息发布for (MqttEndpoint subscriber : subscribers) {subscriber.publish(topic, payload, publish.qosLevel(), publish.isDup(), publish.isRetain());}}}});endpoint.publishAcknowledgeHandler(messageId -> {System.out.println("received ack for message =" + messageId);}).publishReceivedHandler(messageId -> {endpoint.publishRelease(messageId);}).publishCompletionHandler(messageId -> {System.out.println("Received ack for message =" + messageId);});endpoint.publishReleaseHandler(endpoint::publishComplete);}

isTopicMatch方法

用于判断订阅的主题和发布的主题是否匹配。它会将主题按照"/“进行分割，并进行逐层匹配，支持通配符”+“和”#"。

private boolean isTopicMatch(String subscribedTopic, String publishedTopic) {String[] publishTopicArray = publishedTopic.split("/");String[] subscribedTopicArray = subscribedTopic.split("/");//将两个要比较的topic分割//订阅的topic长度不能比发布的topic长一个以上if (subscribedTopicArray.length - 1 > publishTopicArray.length) {return false;}//如果发布的topic长度比订阅的topic长度要长//并且订阅的topic最后不是以#结尾都返回false，因为这不可能if (subscribedTopicArray.length<publishTopicArray.length){if (!subscribedTopicArray[subscribedTopicArray.length-1].equals("#")){return false;}}//对两个topic进行比较for (int i = 0; i < publishTopicArray.length && i < subscribedTopicArray.length; i++) {//如果匹配成功或者匹配到了+，进行下一层匹配if (subscribedTopicArray[i].equals(publishTopicArray[i])||subscribedTopicArray[i].equals("+")){continue;}//如果匹配到了#，直接通过if (subscribedTopicArray[i].equals("#")) {break;}return false;}return true;}

isValidTopic方法

用于判断主题是否合法。它会检查主题是否包含空格，并且要么以"/#“结尾，要么不包含”#"。

public boolean isValidTopic(String topic) {//topic 不能包含任何空格，并且要么以 /# 结尾，要么不包含 #return (!topic.matches(".*\\s+.*"))&&(topic.matches(".*(?:\\/#)?$"));}

完整代码

package com.yichen.starter;import io.netty.handler.codec.mqtt.MqttProperties;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttTopicSubscription;
import io.vertx.mqtt.messages.codes.MqttSubAckReasonCode;import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;/*
* 缺少
* */
public class MqttVertical extends AbstractVerticle {private static Map<String, List<MqttEndpoint>> topicSubscribers;//存储每个topic的订阅关系private static Map<MqttEndpoint, List<String>> subscriptions;//存储每个设备订阅的topic@Overridepublic void start() throws Exception {MqttServer mqttServer = MqttServer.create(vertx);topicSubscribers = new HashMap<>();subscriptions=new HashMap<>();mqttServer.endpointHandler(endpoint -> {//使用lambda表达式实现了endpointHandler方法，传入参数endpoint；System.out.println("MQTT client [" + endpoint.clientIdentifier() + "] request to connect, clean session = " + endpoint.isCleanSession());if (endpoint.auth() != null) {System.out.println("[username = " + endpoint.auth().getUsername() + ", password = " + endpoint.auth().getPassword() + "]");}System.out.println("[properties = " + endpoint.connectProperties() + "]");// accept connection from the remote clientendpoint.accept(false);SubscribeHandle(endpoint);UnSubscribeHandle(endpoint);ReceiveHandle(endpoint);DisConnectHandle(endpoint);});mqttServer.listen(ar -> {if (ar.succeeded()) {System.out.println("MQTT server is listening on port " + ar.result().actualPort());} else {System.out.println("Error on starting the server");ar.cause().printStackTrace();}});}/** 设备断开处理* */private void DisConnectHandle(MqttEndpoint endpoint) {endpoint.disconnectMessageHandler(disconnectMessage -> {System.out.println("Received disconnect from client, reason code = " + disconnectMessage.code());System.out.println("Client "+endpoint.auth().getUsername()+"disconnect with");//对两个映射订阅关系的列表进行更新for (String topic:subscriptions.get(endpoint)){topicSubscribers.get(topic).remove(endpoint);System.out.println("["+topic+"]");}subscriptions.remove(endpoint);});}/**处理订阅消息* */private void SubscribeHandle(MqttEndpoint endpoint) {endpoint.subscribeHandler(subscribe -> {Boolean IsValidTopic=false;//存储订阅消息中要订阅的topic的列表List<MqttTopicSubscription> topicSubscriptions = subscribe.topicSubscriptions();//存储订阅topic Qos级别的列表List<MqttSubAckReasonCode> reasonCodes = new ArrayList<>();//遍历列表for (MqttTopicSubscription s : topicSubscriptions) {//topicString topic = s.topicName();//Qos级别MqttQoS qos = s.qualityOfService();//判断topic是否合法if (!isValidTopic(topic)){//不合法则向设备发送消息endpoint.publish(topic, Buffer.buffer("非法topic，topic不可包含空格"), qos, false, false);continue;}else {IsValidTopic=true;}System.out.println("Subscription for " + topic + " with QoS " + qos);reasonCodes.add(MqttSubAckReasonCode.qosGranted(qos));//判断是否已有此topic，如果有则直接添加，没有则新建键值对if (!topicSubscribers.containsKey(topic)) {topicSubscribers.put(topic, new ArrayList<MqttEndpoint>());}topicSubscribers.get(topic).add(endpoint);//同上if (!subscriptions.containsKey(endpoint)) {subscriptions.put(endpoint, new ArrayList<String>());}subscriptions.get(endpoint).add(topic);}if(IsValidTopic){endpoint.subscribeAcknowledge(subscribe.messageId(), reasonCodes, MqttProperties.NO_PROPERTIES);}});}/** 处理退订* */private void UnSubscribeHandle(MqttEndpoint endpoint) {endpoint.unsubscribeHandler(unsubscribe -> {//遍历要退订的topicfor (String unsubscribedTopic : unsubscribe.topics()) {topicSubscribers.get(unsubscribedTopic).remove(endpoint);//如果某topic的订阅列表为空，删除topicif (topicSubscribers.get(unsubscribedTopic).size() == 0) {topicSubscribers.remove(unsubscribedTopic);}subscriptions.get(endpoint).remove(unsubscribedTopic);//同上if (subscriptions.get(endpoint).size()==0){subscriptions.remove(endpoint);}System.out.println("unsubscribed :" + endpoint.auth().getUsername() + "for" + unsubscribedTopic);}endpoint.unsubscribeAcknowledge(unsubscribe.messageId());});}private void ReceiveHandle(MqttEndpoint endpoint) {endpoint.publishHandler(publish -> {String topic = publish.topicName();Buffer payload = publish.payload();//对topic的合法性进行判断if (!isValidTopic(topic)){endpoint.publish(topic, Buffer.buffer("非法topic，topic不可包含空格"), MqttQoS.AT_MOST_ONCE, false, false);return;}//记录日志接收到设备发布的消息System.out.println("Received message [" + publish.payload().toString(Charset.defaultCharset()) + "] with QoS [" + publish.qosLevel() + "]");if (publish.qosLevel() == MqttQoS.AT_LEAST_ONCE) {endpoint.publishAcknowledge(publish.messageId());} else if (publish.qosLevel() == MqttQoS.EXACTLY_ONCE) {endpoint.publishReceived(publish.messageId());}//遍历订阅关系，进行消息发布for (Map.Entry<String, List<MqttEndpoint>> entry : topicSubscribers.entrySet()) {String subscribedTopic = entry.getKey();//被订阅的topicList<MqttEndpoint> subscribers = entry.getValue();//订阅上方topic的订阅者//判断消息发布的topic是否能和被设备订阅的topic按照规则匹配if (isTopicMatch(subscribedTopic, topic)) {//若匹配，则遍历topic订阅者列表，并进行消息发布for (MqttEndpoint subscriber : subscribers) {subscriber.publish(topic, payload, publish.qosLevel(), publish.isDup(), publish.isRetain());}}}});endpoint.publishAcknowledgeHandler(messageId -> {System.out.println("received ack for message =" + messageId);}).publishReceivedHandler(messageId -> {endpoint.publishRelease(messageId);}).publishCompletionHandler(messageId -> {System.out.println("Received ack for message =" + messageId);});endpoint.publishReleaseHandler(endpoint::publishComplete);}
/*
* 判断topic是否匹配
* */private boolean isTopicMatch(String subscribedTopic, String publishedTopic) {String[] publishTopicArray = publishedTopic.split("/");String[] subscribedTopicArray = subscribedTopic.split("/");//将两个要比较的topic分割//订阅的topic长度不能比发布的topic长一个以上if (subscribedTopicArray.length - 1 > publishTopicArray.length) {return false;}//如果发布的topic长度比订阅的topic长度要长//并且订阅的topic最后不是以#结尾都返回false，因为这不可能if (subscribedTopicArray.length<publishTopicArray.length){if (!subscribedTopicArray[subscribedTopicArray.length-1].equals("#")){return false;}}//对两个topic进行比较for (int i = 0; i < publishTopicArray.length && i < subscribedTopicArray.length; i++) {//如果匹配成功或者匹配到了+，进行下一层匹配if (subscribedTopicArray[i].equals(publishTopicArray[i])||subscribedTopicArray[i].equals("+")){continue;}//如果匹配到了#，直接通过if (subscribedTopicArray[i].equals("#")) {break;}return false;}return true;}public boolean isValidTopic(String topic) {//topic 不能包含任何空格，并且要么以 /# 结尾，要么不包含 #return (!topic.matches(".*\\s+.*"))&&(topic.matches(".*(?:\\/#)?$"));}
}

最后

以上就是我的全部代码和相关内容解释，如果有疑问欢迎评论区交流，如果帮到你了可以点个赞

这篇关于Vert.x框架实现MQTT服务，代码实现及其详解的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/687088。

相关文章

HTML5实现的移动端购物车自动结算功能示例代码

HTML5实现的移动端购物车自动结算功能示例代码

《HTML5实现的移动端购物车自动结算功能示例代码》本文介绍HTML5实现移动端购物车自动结算,通过WebStorage、事件监听、DOM操作等技术,确保实时更新与数据同步,优化性能及无障碍性,提升用... 目录1. 移动端购物车自动结算概述2. 数据存储与状态保存机制2.1 浏览器端的数据存储方式2.1.
阅读更多...
基于 HTML5 Canvas 实现图片旋转与下载功能(完整代码展示)

基于 HTML5 Canvas 实现图片旋转与下载功能(完整代码展示)

《基于HTML5Canvas实现图片旋转与下载功能(完整代码展示)》本文将深入剖析一段基于HTML5Canvas的代码，该代码实现了图片的旋转（90度和180度）以及旋转后图片的下载... 目录一、引言二、html 结构分析三、css 样式分析四、JavaScript 功能实现一、引言在 Web 开发中，
阅读更多...
详解MySQL中DISTINCT去重的核心注意事项

详解MySQL中DISTINCT去重的核心注意事项

《详解MySQL中DISTINCT去重的核心注意事项》为了实现查询不重复的数据,MySQL提供了DISTINCT关键字,它的主要作用就是对数据表中一个或多个字段重复的数据进行过滤,只返回其中的一条数据... 目录DISTINCT 六大注意事项1. 作用范围：所有 SELECT 字段2. NULL 值的特殊处
阅读更多...
SpringBoot中使用Flux实现流式返回的方法小结

SpringBoot中使用Flux实现流式返回的方法小结

《SpringBoot中使用Flux实现流式返回的方法小结》文章介绍流式返回（StreamingResponse）在SpringBoot中通过Flux实现,优势包括提升用户体验、降低内存消耗、支持长连... 目录背景流式返回的核心概念与优势1. 提升用户体验2. 降低内存消耗3. 支持长连接与实时通信在Sp
阅读更多...
Conda虚拟环境的复制和迁移的四种方法实现

Conda虚拟环境的复制和迁移的四种方法实现

《Conda虚拟环境的复制和迁移的四种方法实现》本文主要介绍了Conda虚拟环境的复制和迁移的四种方法实现,包括requirements.txt,environment.yml,conda-pack,... 目录在本机复制Conda虚拟环境相同操作系统之间复制环境方法一：requirements.txt方法
阅读更多...
Spring Boot 实现 IP 限流的原理、实践与利弊解析

Spring Boot 实现 IP 限流的原理、实践与利弊解析

《SpringBoot实现IP限流的原理、实践与利弊解析》在SpringBoot中实现IP限流是一种简单而有效的方式来保障系统的稳定性和可用性,本文给大家介绍SpringBoot实现IP限... 目录一、引言二、IP 限流原理2.1 令牌桶算法2.2 漏桶算法三、使用场景3.1 防止恶意攻击3.2 控制资源
阅读更多...
Python如何去除图片干扰代码示例

Python如何去除图片干扰代码示例

《Python如何去除图片干扰代码示例》图片降噪是一个广泛应用于图像处理的技术,可以提高图像质量和相关应用的效果,：本文主要介绍Python如何去除图片干扰的相关资料,文中通过代码介绍的非常详细,... 目录一、噪声去除1. 高斯噪声（像素值正态分布扰动）2. 椒盐噪声（随机黑白像素点）3. 复杂噪声（如伪
阅读更多...
SQL BETWEEN 语句的基本用法详解

SQL BETWEEN 语句的基本用法详解

《SQLBETWEEN语句的基本用法详解》SQLBETWEEN语句是一个用于在SQL查询中指定查询条件的重要工具,它允许用户指定一个范围,用于筛选符合特定条件的记录,本文将详细介绍BETWEEN语... 目录概述BETWEEN 语句的基本用法BETWEEN 语句的示例示例 1：查询年龄在 20 到 30 岁
阅读更多...
springboot下载接口限速功能实现

springboot下载接口限速功能实现

《springboot下载接口限速功能实现》通过Redis统计并发数动态调整每个用户带宽,核心逻辑为每秒读取并发送限定数据量,防止单用户占用过多资源,确保整体下载均衡且高效,本文给大家介绍spring... 目录 一、整体目标 二、涉及的主要类/方法✅ 三、核心流程图解（简化） 四、关键代码详解1️⃣ 设置
阅读更多...
Java Spring ApplicationEvent 代码示例解析

Java Spring ApplicationEvent 代码示例解析

《JavaSpringApplicationEvent代码示例解析》本文解析了Spring事件机制,涵盖核心概念（发布-订阅/观察者模式）、代码实现（事件定义、发布、监听）及高级应用（异步处理、... 目录一、Spring 事件机制核心概念1. 事件驱动架构模型2. 核心组件二、代码示例解析1. 事件定义
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2