技术答疑.消息队列.DotNetMQ

【DotNetMQ】

基于.NET的开源消息队列框架。

https://www.codeproject.com/Articles/193611/DotNetMQ-A-Complete-Message-Queue-System-for-NET

https://github.com/hikalkan/dotnetmq

【工作原理示意图】

【工作原理介绍】

DotNetMQ提供了3种角色：

客户端（MDSClient）：负责客户端的消息发送和接收。

服务端（MDSServer）：负责服务端的Socket连接管理，消息的接收、发送、分发。

管理端（MDSController）：分别提供了客户端控制器和服务端控制器。客户端控制器可以实时获取服务端的相关运行数据，也可以发送控制指令给服务端。服务端控制器负责接收控制指令，执行控制指令。

【MDSServer】

功能说明：DotNetMQ的服务端，用于管理客户端连接，消息接收、持久化，分发等。

工作原理：

MDSServer一经实例化，就会根据配置文件MDSSettings.xml中配置的Applications，创建MDSClientApplication实例。同时还会创建一个MDSController实例，用于实现MDS管理功能。

MDSServer的启动过程如下（MDSServer.Start）：

<1>使用StorageManager连接并打开数据库。

<2>CommunicationLayer启动通讯层，实现客户端连接的监听及管理（创建连接，注册连接，分配应用程序，监听数据接收）。

客户端与服务端建立Socket长连接的全过程如下：

(1)服务端TCPCommunicationManager启动监听Socket，开启线程监听并接收（TCPConnectionListener.DoListenAsThread）连入到指定端口上的客户端长连接（TCPCommunicator）。

(2)客户端实例化MDSClient，并执行MDSClient.Connect()。

(3)服务端TCPConnectionListener监听到新连接进来后，自动分配一个唯一的ComminicatorId，同时触发TCPConnectionListener的TCPClientConnected事件，在该事件中创建TCPCommunicator，再触发TCPCommunicationManager的CommunicatorConnected事件，该事件由CommunicationLayer的Manager_CommunicatorConnected托管。

在CommunicatorConnected事件中，将新TCPCommunicator加入到通讯层连接集合_communicators中，并立即启动该TCPCommunicator。开启线程，监听消息接收。

(4)客户端连接成功后，立即发送注册消息（MDSRegisterMessage），并等待服务端应答，如果超过30秒无应答，视之为注册失败，连接不成功。

(5)服务端TCPCommunicator接收到数据后，触发消息接收事件MessageReceived，该事件由CommunicationLayer的Communicator_MessageReceived方法托管，此事件只处理客户端的注册消息MDSRegisterMessage，收到客户端注册消息后，将客户端连接加入到客户端应用程序MDSClientApplication的_communicators中，并将该连接的ComminicatorId回发给客户端MDSClient。

(6)客户端收到服务端注册成功的消息后，更新本地TCPChannel的ComminicatorId。

至此客户端和服务端的Socket长连接正式完成握手对接，且都分配了唯一识别码ComminicatorId。

<3>OrganizationLayer启动组织层，实现消息的处理、响应和分发。

<3.1>MDSClientApplication启动客户端应用程序

根据配置文件MDSSettings.xml中配置的Applications，分别启动客户端应用程序，最后再启动一个MDSController程序。

MDSRemoteApplication自行管理TCPCommunicator的接入和关闭。

MDSClientApplication基于MDSRemoteApplication实现，MDSClientApplication的MessageReceived事件由OrganizationLayer的RemoteApplication_MessageReceived方法托管。

具体实现过程如下：

(a)MDSClientApplication的TCPCommunicator接收消息，通过MDSRemoteApplication的Communicator_MessageReceived方法，将消息塞入QueueProcessorThread的队列中。

(b)QueueProcessorThread监听到消息入队后，调用MDSRemoteApplication的IncomingMessageQueue_ProcessItem方法，在该方法中，触发MDSRemoteApplication的MessageReceived事件，该事件由OrganizationLayer的RemoteApplication_MessageReceived方法托管。

(c)RemoteApplication_MessageReceived方法实现对消息的数据持久化和发送回应消息（MDSOperationResultMessage）给客户端。

MDSController的MessageReceived事件由他自己的MDSController_MessageReceived方法托管。

MDSClientApplication实现以下功能：

消息发送者角色（Send_App）：

(a)服务端应用程序等待接收客户端MDSClient发过来的消息。

(b)服务端应用程序将消息转给指定的接收应用程序。

(c)服务端接收应用程序完成消息的数据持久化和内存队列入列。

(d)消息入列后，服务端应用程序发送MDSOperationResultMessage消息通知客户端MDSClient，表示该消息已发送成功。

(e)客户端MDSClient收到服务端回发的消息发送成功的消息后，即表示发送操作完成，否则表示发送操作失败（操作超时、连线被关闭、消息被拒绝、其它未知异常）。

消息接收者角色（Receive_App）：

(a)服务端MDSPersistentRemoteApplicationBase启动线程，通过ProcessWaitingMessageRecordsAsThread方法，从数据库message表或者内存队列_waitingMessages中提取消息，通过MessageDeliverer挑选一个订阅了消息的客户端MDSClient，并将此消息发送给此客户端。

(b)客户端消费完消息后，回发MDSOperationResultMessage消息给服务端。

(c)服务端MDSRemoteApplicat