MQTT---HiveMQ源码详解Netty-MQTT消息事件处理(流程)

Posted 西安-PP

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了MQTT---HiveMQ源码详解Netty-MQTT消息事件处理(流程)相关的知识，希望对你有一定的参考价值。

源博客地址:http://blog.csdn.net/pipinet123

MQTT交流群:221405150

简介

前面这些章节，讲的基本上都是属于netty对MQTT周边的一些处理，由于MQTT协议总共目前可用的消息类型有14个，如果再加上对应的事件处理加载一起那就估计大概有14*3个handler，如果每个来讲一遍，难免有些枯燥，而且知识点会很分散，思考再三，想把整体的MQTT消息以及对应的事件处理作为一节来介绍，我们只讲它整体的实现思路、处理流程即可，这样对需要自己写broker的朋友的帮助应该是非常大的，这也符合最初写此系列博客的初衷。

热身

一、Callback

callback

1、分类

HiveMQ的Callback总体分为同步、异步两种callback，其中部分异步callback被标记为lowlevel。

2、同步

可以看出同步的callback主要分为broker的callback、安全相关的callback、OnConnectCallback、OnPublishReceivedCallback、OnSubscribeCallback，这些回调都是使用异步线程调用。

  1. broker在启动和关闭时，会触发OnBrokerStart和OnBrokerStop，用户可在broker启动的时候做一些自己的处理，例如数据库连接池的创建，spring context的创建等等；在broker关闭时，可以关闭数据库连接池等操作。

  2. 安全相关的主要包括Authentication、Authorization,主要是做连接认证和授权；可以写第三方plugin去做Authentication和Authorization。

  3. OnConnectCallback、OnPublishReceivedCallback、OnSubscribeCallback，用户可以在client连接、client publish、client subscribe的时候做一些处理。

3、异步

  1. 异步callback主要包括一些mqtt消息回调、认证完成回调等等，用户可以根据自己的需求开发一些个性化插件定制属于自己的broker业务。

4、lowlevel

  1. lowlevel属于异步callback的一部分，都是mqtt消息的回调。

5、CallbackExecutor

  1. CallbackExecutor就是所有异步调用callback处理的Executor，由hivemq统一调配；用户可使用配置内部参数来控制其线程数；来保证broker的性能；CallbackExecutor由CallbackExecutorProvider创建提供。

三、Plugin*Handler

在netty handlers一览中介绍了很多plugin*handler；这些handler都是监听netty的用户自定义event来对callback进行回调

正戏

下来就通过mqtt的connect消息的整个调用处理流程来示例一下mqtt消息和事件处理。

Created with Raphaël 2.1.0 MqttConnectHandler MqttConnectHandler MqttConnectHandler MqttConnectHandler PluginOnAuthenticationCallbackHandler PluginOnAuthenticationCallbackHandler PluginOnAuthenticationCallbackHandler PluginOnAuthenticationCallbackHandler PluginAfterLoginCallbackHandler PluginAfterLoginCallbackHandler PluginRestrictionsCallbackHandler PluginRestrictionsCallbackHandler PluginRestrictionsCallbackHandler PluginRestrictionsCallbackHandler PluginOnConnectCallbackHandler PluginOnConnectCallbackHandler ChannelPersistence ChannelPersistence MqttConnectPersistenceHandler MqttConnectPersistenceHandler 当接受到connect消息时 为pipeline添加MqttDisallowSecondConnect(请查看协议 MQTT-3.1.0-2) 验证clientid长度是否符合配置,否则发送ConnAck(REFUSED_IDENTIFIER_REJECTED)到client端 删除IdleStateHandler和NoConnectIdleEventHandler(连接建立后,必须在用户配置时间内发送connect消息) 触发PluginOnAuthentication事件,让其调用callback进行认证 异步遍历所有OnAuthenticationCallback让其认证,每一个callback认证完成会触发一个PluginOnAuthenticationCallbackCompleted事件 接收到PluginOnAuthenticationCallbackCompleted,根据用户的插件认证配置决定下一步处理 当认证完成后会触发PluginOnAuthenticationCompleted 根据client端是否存在LWT,做LWT处理(此处不做过多描述,主要目的是描述流程) 为pipeline添加:PluginAfterLoginCallbackHandler,做认证完成回调处理 触发PluginAfterLogin事件,让其调用callback进行认证完成结果的通知 若认证不通过则发送ConAck(OnAuthenticationCallback返回的return code)到客户端 添加PluginRestrictionsCallbackHandler,为客户端进行授权。 触发PluginRestrictionsAfterLogin事件,让其遍历调用RestrictionsAfterLoginCallback,让每个callback对客户端进行授权 当每个授权都完成后,触发PluginRestrictionsAfterLoginCompleted,将授权信息进行回传 为pipeline添加:PluginOnConnectCallbackHandler,让其遍历所有callback进行连接通知 当所有OnConnectCallback回调完成,触发PluginOnConnectCompleted事件 处理掉与当前clientid一样的连接 保存连接 添加closeFuture,处理客户端断线 若客户端持久session,则触发持久session事件,让MqttConnectPersistenceHandler处理持久session处理 采集(统计)当前在线连接数增加 添加closeFuture,采集(统计)当前在线连接数减少

以上是关于MQTT---HiveMQ源码详解Netty-MQTT消息事件处理(流程)的主要内容，如果未能解决你的问题，请参考以下文章

MQTT---HiveMQ源码详解插件加载

MQTT---HiveMQ源码详解(十三)Netty-MQTT消息事件处理(源码举例解读)

MQTT---HiveMQ源码详解配置加载

MQTT---HiveMQ源码详解(十四)Persistence-LocalPersistence

MQTT---HiveMQ源码详解(十六)TopicTree

MQTT---HiveMQ源码详解Netty-Statistics