简介
前面这些章节，讲的基本上都是属于netty对MQTT周边的一些处理，由于MQTT协议总共目前可用的消息类型有14个，如果再加上对应的事件处理加载一起那就估计大概有14*3个handler，如果每个来讲一遍，难免有些枯燥，而且知识点会很分散，思考再三，想把整体的MQTT消息以及对应的事件处理作为一节来介绍，我们只讲它整体的实现思路、处理流程即可，这样对需要自己写broker的朋友的帮助应该是非常大的，这也符合最初写此系列博客的初衷。
热身
一、Callback
1、分类
HiveMQ的Callback总体分为同步、异步两种callback，其中部分异步callback被标记为lowlevel。
2、同步
可以看出同步的callback主要分为broker的callback、安全相关的callback、OnConnectCallback、OnPublishReceivedCallback、OnSubscribeCallback，这些回调都是使用异步线程调用。
- broker在启动和关闭时，会触发OnBrokerStart和OnBrokerStop，用户可在broker启动的时候做一些自己的处理，例如数据库连接池的创建，spring context的创建等等；在broker关闭时，可以关闭数据库连接池等操作。
- 安全相关的主要包括Authentication、Authorization,主要是做连接认证和授权；可以写第三方plugin去做Authentication和Authorization。
- OnConnectCallback、OnPublishReceivedCallback、OnSubscribeCallback，用户可以在client连接、client publish、client subscribe的时候做一些处理。
3、异步
- 异步callback主要包括一些mqtt消息回调、认证完成回调等等，用户可以根据自己的需求开发一些个性化插件定制属于自己的broker业务。
4、lowlevel
- lowlevel属于异步callback的一部分，都是mqtt消息的回调。
5、CallbackExecutor
- 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.0MqttConnectHandlerMqttConnectHandlerMqttConnectHandlerMqttConnectHandlerPluginOnAuthenticationCallbackHandlerPluginOnAuthenticationCallbackHandlerPluginOnAuthenticationCallbackHandlerPluginOnAuthenticationCallbackHandlerPluginAfterLoginCallbackHandlerPluginAfterLoginCallbackHandlerPluginRestrictionsCallbackHandlerPluginRestrictionsCallbackHandlerPluginRestrictionsCallbackHandlerPluginRestrictionsCallbackHandlerPluginOnConnectCallbackHandlerPluginOnConnectCallbackHandlerChannelPersistenceChannelPersistenceMqttConnectPersistenceHandlerMqttConnectPersistenceHandler当接受到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,采集(统计)当前在线连接数减少
流程较多，部分细节处理做了减免描述，为了让大家更清晰地了解消息的处理、事件的处理、以及其相互之间的触发方式，大家如果自己写broker，没必要生搬硬套按照这样的步骤去处理(只要按照mqtt标准/建议的处理流程去处理即可)，目的是为了大家了解其处理机制提供一个思路而已。
总结
1、所有的mqtt消息的handler与callbackhandler都是通过netty的自定义event来实现交互的，callbackhandler几乎都是动态加入到pipeline中以减少内存的消耗。
2、所由callbackhandler都使用CallbackExecutor去异步调用callback，并监听对应完成的event来进行交互。
由上可以看出，所有流程都是采用异步处理，同时限制Executor线程数来限制异步同时处理过多，使用netty自定义event来达到相互的交互、以及客户端(plugin、mqtt client)感知的同步。