Netty的MQTT协议消息系统的设计与实现.pdf

SCIENCE&TECHNOLOGY INFORMATION科技资讯 2023 NO.17 信 息 与 智 能科技资讯SCIENCE&TECHNOLOGY INFORMATIONNetty的MQTT协议消息系统的设计与实现杨建(江西工程学院 江西新余 338000)摘要：MQTT协议体积较小，耗电量通常较低，因此，可以用极少的代码、带宽为连接远程物联网设备提供可靠的实时消息服务，支持大多数平台，适用于硬件资源受限设备、带宽有限、不可靠网络环境使用。因而，MQTT协议支持QoS、轻量、节省带宽、易于实现等特性已经成为IoT通信标准。该文基于Netty技术，分析IoT MQTT协议消息系统的设计与实现过程，对MQTT协议消息系统的设计上的功能性和非功能性需求做出全面分析，做出总体架构设计，从服务器终端实现、消息代理服务器实现方面研究MQTT协议消息系统的模块功能，最终，对MQTT协议消息系统进行功能与性能测试。关键词：Netty编程框架 MQTT协议 消息系统 集群中图分类号：TP393文献标识码：A 文章编号：1672-3791(2023)17-0039-04Design and Implementation of Nettys MQTT Protocol Message SystemYANG Jian(Jiangxi Institute of Engineering,Xinyu,Jiangxi Province,338000 China)Abstract:MQTT protocol is small in size and generally low in power consumption.Therefore,it can provide reliable real-time message service for connecting remote IoT devices with very little code and bandwidth.It supports most platforms and is suitable for use in hardware resource-constrained devices,bandwidth-constrained and unreliable network environments.Therefore,MQTT protocol supports QoS,lightweight,bandwidth saving,easy to implement and other characteristics,and has become the IoT communication standard.Based on Netty technology,this paper analyzes the design and implementation process of the IoT MQTT protocol message system,comprehensively analyzes the functional and non-functional requirements of the MQTT protocol message system design,and makes the overall architecture design.It studies the module functions of the MQTT protocol message system from the aspects of server terminal implementation and message proxy server implementation.Finally,it tests the function and performance of the MQTT protocol message system.Key Words:Netty programming framework;MQTT protocol;Message system;Colony现阶段，MQTT协议主要有两种类型，一种应用于TCP、IP网络，另一种应用于传感器网络，能够提供有序、无损、双向连接。目前，MQTT协议主要针对M2M、物联网应用开发，在设计初期充分遵循简单、轻量原则，减少对网络带宽与设备资源的实际需求，确保可靠性、保证性地交付。MQTT协议使用发布、订阅消息模式，将有效信息分配给一个或多个接收者，在实现同应用程序解耦的过程中，屏蔽负载内容，现已广泛应用于IoT、Mobile Internet、智能家居、穿戴和办公等智能硬件、车载自组织网络和电力能源等行业。DOI：10.16661/ki.1672-3791.2302-5042-8474作者简介：杨建（1991），男，本科，助教，研究方向为电子工程、软件开发、信息安全。39SCIENCE&TECHNOLOGY INFORMATION科技资讯信 息 与 智 能 2023 NO.17 SCIENCE&TECHNOLOGY INFORMATION科技资讯1 MQTT协议MQTT协议中存在发布者、代理者和订阅者3种角色，发布者负责向代理者发布消息，代理者负责订阅者转发消息，其中，消息发布者、订阅者均为客户端，消息代理者为服务端。MQTT协议报文通常由Fixed header、Variable header与Payload共同组成，主要消息类型有11种。MQTT协议能够提供质量截然不同的3种消息传递服务。在QOS0至多一次服务等级中，MQTT协议消息按照基础TCP、IP网络交付，有可能发生丢失或者产生重复。这种消息传递服务质量等级可以应用在非重要情况下。在QOS1至少一次服务等级中，MQTT协议消息可以确保及时送达，有可能产生重复。在QOS2恰好一次服务等级中，MQTT协议消息可以确保送达且仅被送达一次1。2 MQTT协议消息系统的设计需求分析2.1 功能性需求在MQTT协议消息系统的功能性需求上，主要针对一般用户和系统用户实现。通过对MQTT协议消息系统的设计，IoT物模型管理能够对设备功能实施数字化定义，云端建构实体数据模型后，描述模型功能，便捷化管理设备。系统可以提供IoT设备状态管理，通过上线、下线、离线监控，实时获取设备运行状态、健康状态的数据参数，据此可以搭建设备IoT数据平台，为诊断设备故障、预警夯实基础。端到端的消息路由可以实现对数据的灵活控制，提高边缘计算节点的安全性，便于数据在设备、函数应用与IoT Hub间流转2。在数据、监控状态上报、汇聚到系统中的规则引擎后，根据预先设定规则转发消息至主体设备、业务系统、App、数据库、消息中间件等。系统安全模块通过在设备连接阶段进行设备身份认证，对非法连接请求进行拒绝，确保连接阶段设备合法。与此同时，消息系统通过控制订阅主题进行权限区分，防止因数据泄密、越级访问产生网络阻塞等。为保证MQTT协议消息系统的可用性始终处于高水平，系统需要在集群工作管理模式下，对外部提供一致性服务，确保集群工作过程中的消息发布和消息订阅操作，能够通过正确处理消息的方式，确保消息在各节点的共享。同时，系统可以通过IoT设备调用服务、IoT设备远程完整性监控等通用信息服务能力，支持IoT中心所接收的实时传感器数据进行可视化展示、IoT场景数据采集存储服务、第三方服务等业务相关应用，在系统信息监控过程中，对主题建模最佳数量、序列与角度、CPU使用率、连接总数、活跃数、最大并发数、客户端间发布和订阅信息，监控服务器实时运行状态。2.2 非功能性需求为保证MQTT协议消息系统的实用价值，在非功能性需求上，需要在消息迟延时间、程序并发性和系统可扩展性几方面进行对应约束3。MQTT协议消息系统在传递IoT场景中设备消息方面发挥主要作用，传递消息的及时性，直接决定IoT设备是否能够通过相关指令迅速进行对应操作，因而，需要MQTT协议消息系统在获得消息后，通过端到端的消息路由实时分发消息，系统性能越好。在设计MQTT协议消息系统过程中，需要侧重考虑系统在程序上的并发性，一旦系统所接入的设备数量大规模增长，并发数逐渐增高，需要考虑并发连接支持高并发的连接数，以此保证消息系统的稳定程度。伴随业务扩展、新需求到来，为便于MQTT协议消息系统在设计过程中有效实施扩展，需要通过集群方式对系统进行部署，开发新功能、提供统一接口。3 MQTT协议消息系统的总体架构设计3.1 Netty技术解析Netty是经由JBOSS应用服务器所提供的Java网络开源应用框架，功能强大。Netty提供全新方式开发Web server、客户端过程，该方式易用性、扩展性显著增强，在复杂的内部环境基础上，允许更高的吞吐量。一以贯之，Netty能够提供异步通信方式、事件驱动策略的Web应用程序框架与工具，快速实现Web server、客户端管理程序的高性能、高可靠性开发。Netty使用传输控制协议，为多客户端提供消息查询服务，促使该过程相对简单。现阶段，Netty的核心组件主要包括由java.nio.channels包定义的Channel通道、Callback回调、异步模型Future、可扩展的事件模型和Netty、代码的主要扩展与定制点Channel Handler4。其中，Channel通道通常代表一个实体的开放性连接，异步模型Future提供在操作完成过程中通知应用程序的方式，其能够在未来某一时刻完成，提供对结果的访问，Channel Handler是一个父接口，本身并未提供过多方法。3.2 MQTT协议消息系统的总体设计现阶段，MQTT协议消息系统是IoT重要的传输协议，应用广泛，负责进行设备接入和采集数据，需要根据业务实际需求将获取的数据存储至Mysql数据库、40SCIENCE&TECHNOLOGY INFORMATION科技资讯 2023 NO.17 信 息 与 智 能科技资讯SCIENCE&TECHNOLOGY INFORMATION统一上报到相关业务系统等。在搭建MQTT协议消息系统过程中主要以集群方式实现，通过集群的方式可以解决基于MQTT协议的Rabbit MQ消息收发，实现远程收发、AMQP与MQTT间的收发。单个节点能够提供完整服务，可以支撑MQTT协议消息的传递，多节点协调对外提供一致性服务。每单个节点的MQTT代理服务端在搭建过程中，使用Netty技术建构Boss Event Loop线程模型，用以处理客户端连接请求，建构Work Event Loop线程模型，用以处理客户端的读写请求。消息入站后进行字节解码，消息出站前将MQTT客户端对象编译成为字节5。Engine x是高性能Web服务器，其作为云上数据库的存储引擎，可以通过数据流代理Stream模块实现协议的代理转发tcp报文，主要用于数据流代理和负载均衡等。分布式应用程序协调服务软件zookeeper作为服务集群，MQTT协议消息系统的总体设计主要由传感器、车载装置、智慧工厂等物模型与各个主题模块、设备运行状态管理模块、消息路由模块、安全模块（SE）、集群模块、监控模块组成。4 MQTT协议消息系统的模块功能实现4.1 服务器终端实现服务器终端能够在消息系统中接受、发送、解析和计算IoT设备控制命令与数据指令，全监控IoT设备与数据存储设备信息，主要在下述几个模块进行。模块一，传感器、车载装置、智慧工厂等物模型与各个主题模块。消息系统通过TSL语言描述物模型，采取JSON格式对物模型数据进行类似格式上报，在主题实现过程中将IoT设备主题划分为上行、下行两类数据。模块二，设备运行状态管理模块。该模块是后续消息发布和订阅操作实现的基础，需要在连接设备时通过发送报文连接消息系统。消息系统通过定义Session信息实现持久和非持久会话的存储6。一旦设备成功与消息系统建立连接，通过维护TCP长连接，检测设备运行状态，及时将不够活跃的连接断开，充分将服务器内存资源释放出来。基于MQTT协议规范，始终保持连接的设备需要主动向服务节点定时传送PING包REQ心跳请求，系统接收请求后及时进行将心跳响应（PINGRESP）返回至客户端，进而保持设备连接状态，成功测试网络处于通顺状态。为实现设备心跳，需要使用Netty技术实现连接读写事件、空闲事件检查，根据设备心跳时间触发Netty核心组件，找到设备对应连接并予以关闭，释放服务器资源。模块三，消息路由模块。在实现即时消息推送过程中，通过成功连接设备端、系统，首先发送报文请求，进行协议处理，提取消息发布主题、消息传递服务质量级别、消息内容，进行封装后调用发布权限，验证设备是否具备。在消息推送前需要进行主题订阅匹配查询，找出各层级设备订阅列表和订阅该主题消息的设备，写入消息到设备对应消息队列后，将消息推送给指定客户端。在实现离线消息推送过程中，需要结合MQTT协议规范存储设备离线消息。在消息推送后进行主题订阅匹配查询，通过设备状态查询模块查找设备在线情况，便于系统通过离线数据写入模块，将设备离线消息写入数据库7。一旦系统初始化开始，为保证离线消息毫秒级查询，需要在处理逻辑中调用generate Row Key生成消息的行键。最终，将消息存储在列表里批量提交，通过后台程序及时删除推送过的消息。在桥接数据到Kafka开源流处理平台时，需要根据预设规则将主题消息筛选，通过写入对应主题，促使其他业务订阅平台主题消费消息。模块四，安全模块（SE）。使用“一型号一密钥”的方式验证设备注册情况，使用一机器一密钥的方式验证设备认证情况。设备连接时，通过提取clientid、username和password这3个参数解析连接报文，向设备身份验证服务发送请求。使用标识、IP、用户名、密码和主题为参数发起请求权限，处理ACL HTTP请求。模块五，集群模块。集群通常由节点构成，节点不规则分布于JVM、物理机之上，一旦集群系统被触发，节点将自动加入集群，节点状态变化通过gossip通信协议传输至集群每个节点，确保集群状态一致性8。因MQTT协议消息系统通过去中心化集群方式搭建，存在多个服务节点，IoT设备发布消息后，节点需要精准转发消息，确保消息路由推送消息的正确与完整程度。模块六，监控模块。该模块通过变量方式存储各节点设备连接数、主题订阅数、客户端等信息，在数据库中，一旦需要查看相应信息，可以依靠脚本及时获取信息并定时更新。4.2 消息代理服务器实现在通信过程，需要设备和服务器间的通信协议、服务器与客户端间利用AMQPAMQP传输的通信协议、服务器与移动通信终端间利用MQTT协议传输的通信协议9。实现MQTT协议消息系统代理服务器工作，需要41SCIENCE&TECHNOLOGY INFORMATION科技资讯信 息 与 智 能 2023 NO.17 SCIENCE&TECHNOLOGY INFORMATION科技资讯科学管理已经订阅的消息，通过多终端转发消息，查询存储信息，同时，通过设置用户权限实现“1&X”服务关系。在多终端服务端口，用户可以通过ID查询订阅成功的MQTT 主题，通过多终端对设备实施远程控制。5 MQTT协议消息系统的功能与性能测试Netty的MQTT协议消息系统的设计与实现，使用客户端工具进行消息系统的功能测试。在测试过程中，客户端工具通过创建多样化连接设置的客户端满足测试需求，选用 MQTT Box 作为主测试工具，验证MQTT协议消息系统是否满足系统功能需求。在功能测试过程中，主要对IoT设备身份认证ID、发布消息、消息订阅主题（Subscribe）、接收推送消息、自定义心跳包结构体、消息重投功能（republish）进行10。针对发布消息、消息订阅主题、接收推送消息、自定义心跳包结构体、消息重投功能的测试通常由用例编号、用例名称、用例标题、测试目的、测试步骤（预置条件、测试步骤）、预期结果、测试结果等共同组成。选用 MQTT.fx，使用 Java 语言编写的客户端工具，针对MQTT客户端进行Topic订阅和消息发布，同时启动客户端A和客户端B，A用于消息发布，B用于Topic 订阅，测试MQTT协议消息系统是否支持QOS2恰好一次服务等级质量11。在单节点性能测试上，可以通过两台机器执行程序模拟，多次执行命令设置连接数，通过Web监控页面连接成功率，5 000 连接数的 CPU 占用率为 10.2%，10 000连接数的CPU占用率为18.5%，20 000连接数的CPU占用率为28.3%，30 000连接数的CPU占用率为41.6%，38 000连接数的CPU占用率为55.1%，结果均为成功。可见连接数在38 000内，设备连接较为稳定。在集群性能测试过程中，通过Jmeter测试软件模拟设置并发连接数及相关参数、点击执行，生成测试报告。MQTT协议消息系统在集群环境下可以处理更多设备连接数，节点集群处理100 000连接非常稳定，CPU负载较低，消息时延在1 s内，满足对系统需求。6 结语综上所述，Netty提供API接口从网络处理代码中解耦核心业务逻辑与辅助功能，完全基于Java NIO（无阻塞的输入/输出）中的Buffer实现内部可扩展性解决方案，在NIO通道进行交互的过程中，将数据移进移出通道，架构编程服务器与客户端框架结构，实现服务端、客户端要求的功能。以Netty技术为支撑的MQTT协议消息系统属于IoT服务运营平台组成部分之一，与IoT设备管理子系统同处于IoT服务运营平台接入层底部，作为平台重要的基础，MQTT协议消息系统可以为平台上层应用提供强大的数据支撑，与接入层的规则引擎、IoT应用微服务、数据存储管理系统通过HTTP API网关、RPC服务器等实现服务间的协同。测试结果表明，MQTT协议消息系统能够满足预期设计目标，可利用在多种物联网系统中。参考文献1 陈文艺,高婧,杨辉.基于MQTT协议的物联网通信系统设计与实现J.西安邮电大学学报,2020(3):26-32.2 陈功平,王红.基于雾计算及RoF-DAS架构的物联网调度系统设计J.太原学院学报(自然科学版),2022(1):53-58.3 孙握瑜.基于NB-IoT技术的物联网网关多协议转换研究J.齐齐哈尔大学学报(自然科学版),2021(4):49-53.4 张娜,史佳炳,吴彪,等.基于Netty和Kafka的IOT终端服务系统设计方案J.浙江理工大学学报(自然科学版),2020(2):240-245.5 赵薇,邓夏唯,付睿翎,等.一种基于JSON格式的网络信息安全系统数据传输方法J.中国传媒大学学报(自然科学版),2020(3):22-28.6 郑应亨,于磊,袁应红.基于物联网的智能建筑安全防范评价研究J.重庆科技学院学报(自然科学版),2022(5):111-116.7 胡荣磊,陈雷,王乐胥.一种支持多类型无线连接方式的安全物联网网关设计J.北京电子科技学院学报,2021(3):48-56.8 吴启保,李金玉,俞龙,等.基于LoRa与MQTT通信的茶园环境信息智能监测系统J.广州航海学院学报,2020(2):55-59,69.9 程仲汉,郑清安,陈淑珍.一种基于Kubernetes的Web应用部署与配置系统J.成都信息工程大学学报,2021(5):503-507.10 孙龙,温建新,马凯.基于海洋网络信息体系的自适应接口技术J.中国电子科学研究院学报，2021,16(9):884-890,905.11 严清虎,龚家元,汪雪涛.基于阿里云物联网平台的无线风速监测系统设计J.湖北汽车工业学院学报,2022(3):46-50.42