activemq-broker接收消息

配置broker时，都会设置connector，connector内部会持有一个TransportServer，TransportServer相当于socketserver，transport相当于socket，connector启动（start）时会引发socketserver的启动（start），TcpTransportServer的主要职责就是accept（socketserver的主要职责就是accept）

//org.apache.activemq.transport.tcp.TcpTransportServer的start
protected void doStart() throws Exception {
  if (useQueueForAccept) {
    Runnable run = new Runnable() {
      @Override
      public void run() {
        try {
            //socketQueue是一个阻塞队列，不断的从socketQueue中获得socket，然后执行handleSocket方法
          while (!isStopped() && !isStopping()) {
            Socket sock = socketQueue.poll(1, TimeUnit.SECONDS);
            if (sock != null) {
              try {
                  //handleSocket最终会调用getAcceptListener().onAccept(configuredTransport)
                handleSocket(sock);
              } catch (Throwable thrown) {
                if (!isStopping()) {
                  onAcceptError(new Exception(thrown));
                } else if (!isStopped()) {
                  LOG.warn("Unexpected error thrown during accept handling: ", thrown);
                  onAcceptError(new Exception(thrown));
                }
              }
            }
          }

        } catch (InterruptedException e) {
          if (!isStopped() || !isStopping()) {
            LOG.info("socketQueue interrupted - stopping");
            onAcceptError(e);
          }
        }
      }
    };
    socketHandlerThread = new Thread(null, run, "ActiveMQ Transport Server Thread Handler: " + toString(), getStackSize());
    socketHandlerThread.setDaemon(true);
    socketHandlerThread.setPriority(ThreadPriorities.BROKER_MANAGEMENT - 1);
    //启动线程，该线程循环处理accept到的socket
    socketHandlerThread.start();
  }
  //TcpTransportServer继承了TransportServerThreadSupport，其本身也持有持有一个线程，
  //该线程用来执行的run逻辑，TcpTransportServer的run逻辑就是accept。
  super.doStart();
}

TcpTransportServer启动后执行的run方法

/**
* 就是accept
* pull Sockets from the ServerSocket
*/
@Override
public void run() {
  if (!isStopped() && !isStopping()) {
    final ServerSocket serverSocket = this.serverSocket;
    if (serverSocket == null) {
      onAcceptError(new IOException("Server started without a valid ServerSocket"));
    }
        //channel不为null说明是nio
    final ServerSocketChannel channel = serverSocket.getChannel();
    if (channel != null) {
        //nio
      doRunWithServerSocketChannel(channel);
    } else {
        //以bio为例
      doRunWithServerSocket(serverSocket);
    }
  }
}

TcpTransportServer的run方法就是accept

private void doRunWithServerSocket(final ServerSocket serverSocket) {
  while (!isStopped()) {
      //accept
    Socket socket = serverSocket.accept();
    if (socket != null) {
      if (isStopped() || getAcceptListener() == null) {
        socket.close();
      } else {
          //默认useQueueForAccept为true，socketQueue中的Socket最终会由socketHandlerThread取出并执行handleSocket(socket)，
          //handleSocket最终会调用getAcceptListener().onAccept(configuredTransport)。
        if (useQueueForAccept) {
          socketQueue.put(socket);
        } else {
            //handleSocket最终会调用getAcceptListener().onAccept(configuredTransport)
          handleSocket(socket);
        }
      }
    }
  }
}

在org.apache.activemq.broker.TransportConnector启动时，设置了AcceptListener（getServer().setAcceptListener）该AcceptListener的主要逻辑：

brokerService.getTaskRunnerFactory().execute(new Runnable() {
  @Override
  public void run() {
    try {
      if (!brokerService.isStopping()) {
          //transport中含有Socket，transport代表client
        Connection connection = createConnection(transport);
        //connection.start会执行transport.start，TcpTransport继承了TransportServerThreadSupport，
        //start会开启该线程的执行逻辑，即TcpTransport的run方法，阻塞读取socket数据。
        connection.start();
      } else {
        throw new BrokerStoppedException("Broker " + brokerService + " is being stopped");
      }
    } catch (Exception e) {
      String remoteHost = transport.getRemoteAddress();
      ServiceSupport.dispose(transport);
      onAcceptError(e, remoteHost);
    }
  }
});

accept到的socket被包装为TcpTransport，并start该Transport，socket就可以开始读数据了

//TcpTransport的run方法会调用doRun
protected void doRun() throws IOException {
  try {
      //从Socket读数据
    Object command = readCommand();
    doConsume(command);//执行命令，比如如果该command是一个Message，则会将其放入对应的destination
  } catch (SocketTimeoutException e) {
  } catch (InterruptedIOException e) {
  }
}

小结：connector.start引发了TransportServer的start，TransportServer的start主要就是socketserver.accept（在独立线程中），accept到的socket可以放入队列异步处理，也可以同步处理（在accept所在线程中），二者的执行逻辑是一致的，都会将socket封装成transport，再将transport封装成connection后进行connection.start，这就开始阻塞读了（transport会持有一个独立线程，阻塞读就在该线程中）。

TcpTransport的读线程会阻塞在readCommand上，当client有数据传来时，readCommand会读取数据，并将其unmarshal为activemq的命令对象，然后再doConsume该对象。doConsume最终会调用broker（TransportConnection持有broker）的相关方法（比如，如果command是Message，就会调用broker.send(producerExchange, messageSend);），这就将Message交由broker处理了（接收client发来的命令并不属于broker的职责，broker真正要做的是将处理这些命令，比如将消息路由置对应的destination，而接收client命令的任务是由TransportServer完成的）。