activeMQ支持多种通讯协议TCP/UDP等,我们选取最常用的TCP来分析activeMQ的通讯机制。首先我们来明确一个概念:
客户(Client):消息的生产者、消费者对activeMQ来说都叫作客户。
消息中转器(Message broker):它是activeMQ的核心,它接收信息并进行相关处理后分发给消息消费者。
为了能清楚的描述出activeMQ的核心通讯机制,我们选择3个部分来进行说明,它们分别是建立链接、关闭链接、心跳。
一、Client跟activeMQ的TCP通讯的初始化过程分析如下:
1. activeMQ初始化时,通过TcpTransportServer类根据配置打开TCP侦听端口,客户通过该端口发起建立链接的动作。
2. 把accept的Socket放入阻塞队列中。
3. 另外一个线程Socket handler阻塞着等待队列中是否有新的Socket,如果有则取出来。
4. 生成一个TransportConnection的实例。TransportConnection类的主要作用是处理链路的状态信息,并实现CommandVisitor接口来完成各类消息的处理。
5. TransportConnection会使用一个由多个TransportFilter实例组成的消息处理链条,负责对接收到的各类消息进行处理并发送相应的应答。这个链条的典型组成顺序:MutexTransport->WireFormatNegotiator->InactivityMonitor->TcpTransport。在这条链条中最后的一环就是TcpTransport类,它是实际和Client获取和发送数据的地方,该类的重要方法有run()和oneway(),一个负责读取,一个负责发送。
6. 建链完成,可以进行通讯操作。
二、关闭链接
activeMQ发现TCP链接的关闭,最关键的代码在TcpBufferedInputStream类中的
int n = in.read(buffer, position, buffer.length - position);
三、心跳
为了更好的维护TCP链路的使用,activeMQ采用了心跳机制作为判断双方链路的健康情况。activeMQ使用的是双向心跳,也就是activeMQ的Broker和Client双方都进行相互心跳,但不管是Broker或Client心跳的具体处理情况是完全一样的,都在InactivityMonitor类中实现,下面具体介绍。
心跳会产生两个线程“InactivityMonitor ReadCheck”和“InactivityMonitor WriteCheck”,它们都是Timer类型,都会隔一段固定时间被调用一次。ReadCheck线程主要调用的方法是readCheck(),当在等待时间内,有消息接收到,则该方法会返回true。WriteCheck线程主要调用的方法是writeCheck(),这有个小技巧,大家可以参考一下,那就是当WriteCheck线程休眠时,有任何数据发送成功,则该线程被唤醒后,不用通过TCP向对方真的发送心跳消息,这样可以从一定程度上减少网络传输的数据量。
activeMQ的通讯处理机制还有很多,如断开重连等等,非常希望大家可以有兴趣来一起补充。
后文待续。