ActiveMQ之 TCP通讯机制

Posted 2020-06-13

　　

　　ActiveMQ支持多种通讯协议TCP/UDP等，我们选取最常用的TCP来分析ActiveMQ的通讯机制。首先我们来明确一个概念: 
　　客户(Client)：消息的生产者、消费者对ActiveMQ来说都叫作客户。 
　　消息中介(Message broker)：接收消息并进行相关处理后分发给消息的消费者.

　　

 　　　　

 

　　为了能清楚的描述出ActiveMQ的核心通讯机制，我们选择3个部分来进行说明，它们分别是建立链接、关闭链接、心跳。 

一、Client跟activeMQ的TCP通讯的初始化过程分析如下： 
　　(1) ActiveMQ初始化时，通过TcpTransportServer类根据配置打开TCP侦听端口，客户端通过该端口发起建立链接的动作。 
　　(2) 把接收到的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向对方真的发送心跳消息，这样可以从一定程度上减少网络传输的数据量。