ActiveMQ源码架构解析第一节

 

第一篇文章我们先从hello world写起，下面是使用java代码调用activemq的api发送一条消息。

 

public class test {

    public static void main(String[] args) throws Exception {

       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(

              "tcp://localhost:61616");

       Connection connection = connectionFactory.createConnection(); 

       connection.start();      

       Session session = connection.createSession(false,

              Session.AUTO_ACKNOWLEDGE);

       MessageProducer producer = session.createProducer(session

              .createQueue("testQueue"));

       Message message = session.createTextMessage("hello everybody!");

       producer.send(message);

       producer.close();

       session.close();

       connection.close();

    }

}

 

 

 

下面来分析发送消息的几个步骤，如下：

 

1、创建ActiveMQConnectionFactory，入参是url，指定schema以及要连接的ip和端口号，其次是创建ActiveMQConnection，tcp协议交互肯定是要使用Socket类，所以说明下ActiveMQConnection->Transport->Socket的关系，Transport是对Socket的封装，而ActiveMQConnection则是对Transport的封装，如下图所示：

 

2、创建ActiveMQConnection，因为上图已经说明ActiveMQConnection和Transport是组合关系，所以创建ActiveMQConnection时首先要创建Transport，因为ActiveMQ的交互方式分为Tcp、Udp以及HTTP协议，ActiveMQ使用了非常经典的简单工厂设计模式，使用这个模式的好处是工厂可以根据uri的schema头来动态创建相应的TransportFactory工厂，例如用户输入tcp://localhost:61616，ObjectFactory则可以获取到schema是tcp然后来实例化TcpTransportFactory，然后在调用TcpTransportFactory工厂来生产TcpTransport对象，简单工厂模式如下图，我是把2个工厂画到了一起：

         创建完TcpTransport还不够，因为TcpTransport只实现了发送和接受消息，还需要做一些封装来实现相应的业务处理，说到封装这里使用到了包装设计模式，也叫装饰者模式，jdk的java.io输入输出流也有使用，类图如下：
  

1、MutexTransportFilter类实现了对每个请求的同步锁，同一时间只允许发送一个请求，如果有第二个请求需要等待第一个请求发送完毕才可继续发送。

 

2、WireFormatNegotiator类实现了在客户端连接broker的时候先发送数据解析相关的协议信息，例如解析版本号，是否使用缓存等信息。

 

3、InactivityMonitor类实现了连接成功后启动心跳检查机制，客户端每10秒发送一次心跳信息，服务端每30秒读一次心跳信息，如果没有读到则会断开连接，心跳检测是相互的，客户端也会每30秒读取服务端发送来的心跳信息，如果没有读到也一样会断开连接。

 

4、ResponseCorrelator类实现了异步请求但需要获取响应信息否则就会阻塞等待功能。

 

创建ActiveMQConnection的时序图如下：

 

 

客户端调用createConnection()方法，通过TcpTransportFactory获取到TcpTransport对象，得到TcpTransport对象后，TcpTransportFactory又调用自己的configure方法对TcpTransport进行了包装，代码如下：

 

public Transport configure(Transport transport, WireFormat wf, Map options) throws Exception {

    transport = compositeConfigure(transport, wf, options);

    transport = new MutexTransport(transport);

    transport = new ResponseCorrelator(transport);

    return transport;

}

 

这就是包装设计模式的使用，当调用transport.request()发送消息时，时序图如下：

 

       每次transport调用request方法时都会先判断next是否为空，如果不为空则先调用next.request()，当然在调用之前和之后都会加入相应的逻辑，包装设计模式的好处就是当我不想启动心跳检测功能时则可以非常简单的实现，只需要不为WireFormatNegotiator设置next属性为InactivityMonitor即可，获取加入其它功能不需要修改原来的设计只需要新增一个类即可，充分的体现了设计模式的开放封闭原则以及单一职责原则，设计模式在ActiveMQ中可以说是使用的淋漓尽致，恰到好处。

 

    创建完ActiveMQConnection之后就是创建ActiveMQSession以及ActiveMQMessageProducer了，这两个对象的创建就比较简单了没什么可说的，当然在创建这些对象的时候客户端会发送相应的信息给服务端，本节主要讲解连接的建立。信息的交互以及Message的发送就留到下一节在说了，第一次写这么长的博文，可能有的地方讲的思路不清晰，还请大家多多指正，谢谢大家，睡觉了，明天还要加班！

 

评论

1 楼 The_Sword_Best 2015-12-25  

最近也在读activemq的源码，你分析的很到位，拜读了。想问一下你分析的这份源码是哪个版本的？