消息中间件 activeMQ的源码分析 之 TCP通讯机制

  
    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的通讯处理机制还有很多，如断开重连等等，非常希望大家可以有兴趣来一起补充。
后文待续。

 

评论

4 楼 DEMONU 2015-06-08  

这个双向重连有什么好处和弊端了？感觉没必要啊

3 楼 wxq276 2009-11-16  

ActiveMQ的UDP协议怎么用的呀，这两天试了一下，接收时出现警告：

2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 6 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 7 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 8 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 9 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 10 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 11 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 12 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 13 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 6 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 7 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 8 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 9 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 10 is no longer present
2009-11-16 16:07:33 org.apache.activemq.transport.udp.CommandDatagramChannel sendBuffer
警告: Request for buffer: 11 is no longer present

2 楼 boxer1003 2009-07-13  

我只能通过“顶”来支持楼主了，一直想找这方面的资料，谢谢lz大牛的辛勤劳动

1 楼 coofucoo 2009-07-13  

非常不错！