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  7   } 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完成的）。