rabbitmq-connection

一、ConnectionFactory

ConnectionFactory factory = new ConnectionFactory();

 

开启一个连接，默认localhost和默认端口 

factory.newConnection(addressAry);

 

二、newConnection

 

完整的方法签名：

public Connection newConnection(ExecutorService executor, Address[] addrs) throws IOException, TimeoutException

 

作用：

创建一个broker连接。这里的地址连接信息是个数组，在处理时会使用第一个连上的地址。

创建连接时，可以选择automatic connection recovery。如果这个属性设为true，那么当连接断开时，框架会自动连上，此时选择的地址就是随机的。

 

执行步骤：

1、创建FrameHandlerFactory，之后会通过socket发送header消息。

2、创建ConnectionParams，封装一些参数为实体。

3、是否设置isAutomaticRecoveryEnabled

4、如果设置了，则返回AutorecoveringConnection

5、如果没有设置，则返回AMQConnection

 

三、返回AMQConnection步骤

1、FrameHandler

 FrameHandler handler = fhFactory.create(addr);

        1、通过SocketFactory初始化一个socket连接，并封装此socket的input和output流到SocketFrameHandler（extends FrameHandler）。

        2、设置此socket不延时属性。这是一个算法，算法名称：Nagle's algorithm。

             // disable Nagle's algorithm, for more consistently low latency
             socket.setTcpNoDelay(true);

 

2、AMQConnetion

AMQConnection conn = new AMQConnection(params, handler);

conn.start();

      1、initializeConsumerWorkService

           this._workService  = new ConsumerWorkService(executor, threadFactory, shutdownTimeout);

           参数定义：

           public static final int HANDSHAKE_TIMEOUT = 10000;
           private final ExecutorService executor;
           private Thread mainLoopThread;
           private ThreadFactory threadFactory = Executors.defaultThreadFactory();

           构造方法：

           public ConsumerWorkService(ExecutorService executor, ThreadFactory threadFactory, int shutdownTimeout) {
                this.privateExecutor = (executor == null);     //是否是自定义的线程池，如果executor为空，则框架会默认初始化

               // DEFAULT_NUM_THREADS默认为操作系统处理器个数的2倍
                this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS, threadFactory) : executor;
               this.workPool = new WorkPool<Channel, Runnable>();
               this.shutdownTimeout = shutdownTimeout;
          }

 

     2、initializeHeartbeatSender

         

     3、this._running = true; //处于运行中状态

 

     4、_frameHandler.setTimeout(HANDSHAKE_TIMEOUT);
          _frameHandler.sendHeader();

           通过socket发送0-9-1 AMQP版本header信息

 

    5、启动mainloop MainLoop loop = new MainLoop();
        final String name = "AMQP Connection " + getHostAddress() + ":" + getPort();
        mainLoopThread = Environment.newThread(threadFactory, loop, name);
        mainLoopThread.start();

        作用：

        只要_running=true，则一直通过frameHandler循环读取frame。