Kafka源码之SocketServer

线程模型：

1. 当前broker上配置了多少个listener，就有多少个Acceptor，⽤于新建连接。

2. 每个Acceptor对应N个线程的处理器（Processor），⽤于接收客户端请求。

3. 处理器们对应M个线程的处理程序（Handler），处理⽤户请求，并将响应发送给等待给客户写响应的处理器线程。

在启动KakfaServer的startup⽅法中启动SocketServer。

每个listener就是⼀个端点，每个端点创建多个处理程序。

究竟启动多少个处理程序？

processor个数为numProcessorThreads个。上图中for循环为从processorBeginIndex到prodessorEndIndex（不包括）。

numProcessorThread为：

acceptor的启动过程。

KafkaThread。

调⽤Thread的构造器。

KafkaThread的start⽅法即是Thread的start⽅法，此时调⽤的是acceptor的run⽅法：

/**
 * 使⽤Java的NIO
 * 循环检查是否有新的连接尝试
 * 轮询的⽅式将请求交给各个processor来处理。
 */
def run() {
  serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
  startupComplete()
  try {
    var currentProcessor = 0
    while (isRunning) {
      try {
        val ready = nioSelector.select(500)
        if (ready > 0) {
          val keys = nioSelector.selectedKeys()
          val iter = keys.iterator()
          while (iter.hasNext && isRunning) {
            try {
              val key = iter.next
              iter.remove()
              if (key.isAcceptable)
                // 指定⼀个processor处理请求
                accept(key, processors(currentProcessor))
              else
                throw new IllegalStateException("Unrecognized key state for acceptor thread.")
              // round robin to the next processor thread
              // 通过轮询的⽅式找到下⼀个processor线程
              currentProcessor = (currentProcessor + 1) % processors.length
            } catch {
              case e: Throwable => error("Error while accepting connection", e)
            }
          }
        }
      }
      catch {
        // We catch all the throwables to prevent the acceptor thread from exiting on exceptions due
        // to a select operation on a specific channel or a bad request. We don't want
        // the broker to stop responding to requests from other clients in these scenarios.
        case e: ControlThrowable => throw e
        case e: Throwable => error("Error occurred", e)
      }
    }
  } finally {
    debug("Closing server socket and selector.")
    swallowError(serverChannel.close())
    swallowError(nioSelector.close())
    shutdownComplete()
  }
}

Acceptor建⽴连接，处理请求：

/*
 * Accept a new connection
 * 建⽴⼀个新连接
 */
def accept(key: SelectionKey, processor: Processor) {
  // 服务端
  val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
  // 客户端
  val socketChannel = serverSocketChannel.accept()
  try {
    connectionQuotas.inc(socketChannel.socket().getInetAddress)
    // ⾮阻塞
    socketChannel.configureBlocking(false)
    socketChannel.socket().setTcpNoDelay(true)
    socketChannel.socket().setKeepAlive(true)
    // 设置发送缓冲⼤⼩
    if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
      socketChannel.socket().setSendBufferSize(sendBufferSize)
    debug("Accepted connection from %s on %s and assigned it to processor %d,
      sendBufferSize [actual|requested]: [%d|%d] recvBufferSize [actual|requested]: [%d|%d]".format(socketChannel.socket.getRemoteSocketAddress,
      socketChannel.socket.getLocalSocketAddress, processor.id,
      socketChannel.socket.getSendBufferSize, sendBufferSize,
      socketChannel.socket.getReceiveBufferSize, recvBufferSize))
    // 调⽤Processor的accept⽅法，由processor处理请求
    processor.accept(socketChannel)
  } catch {
    case e: TooManyConnectionsException =>
      info("Rejected connection from %s, address already has the configured maximum of %d connections.".format(e.ip, e.count))
    close(socketChannel)
  }
}

Processor将连接加⼊缓冲队列，同时唤醒处理线程。

Processor的run⽅法从newConnections中取出请求的channel，解析封装请求，交给handler处理。

将请求信息放到请求队列中。

在KafkaServer的startup⽅法中实例化KafkaRequestHandlerPool，该类会⽴即初始化numIoThreads个线程⽤于执⾏KafkaRequestHandler处理请求的逻辑。

KafkaRequestHandlerPool以多线程的⽅式启动多个KafkaRequestHandler。

KafkaRequestHandler的run⽅法中，receiveRequest⽅法从请求队列获取请求。

KafkaRequestHandler的run⽅法中使⽤模式匹配。

上图中，apis的handle⽅法处理请求。

/**
 * 处理所有请求的顶级⽅法，使⽤模式匹配，交给具体的api来处理
 */
def handle(request: RequestChannel.Request) {
  try {
    trace(s"Handling request:${request.requestDesc(true)} from connection ${request.context.connectionId};" +
      s"securityProtocol:${request.context.securityProtocol},principal:${request.context.principal}")
    request.header.apiKey match {
      case ApiKeys.PRODUCE => handleProduceRequest(request)
      case ApiKeys.FETCH => handleFetchRequest(request)
      case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)
      case ApiKeys.METADATA => handleTopicMetadataRequest(request)
      case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)
      case ApiKeys.STOP_REPLICA => handleStopReplicaRequest(request)
      case ApiKeys.UPDATE_METADATA => handleUpdateMetadataRequest(request)
      case ApiKeys.CONTROLLED_SHUTDOWN => handleControlledShutdownRequest(request)
      case ApiKeys.OFFSET_COMMIT => handleOffsetCommitRequest(request)
      case ApiKeys.OFFSET_FETCH => handleOffsetFetchRequest(request)
      case ApiKeys.FIND_COORDINATOR => handleFindCoordinatorRequest(request)
      case ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request)
      case ApiKeys.HEARTBEAT => handleHeartbeatRequest(request)
      case ApiKeys.LEAVE_GROUP => handleLeaveGroupRequest(request)
      case ApiKeys.SYNC_GROUP => handleSyncGroupRequest(request)
      case ApiKeys.DESCRIBE_GROUPS => handleDescribeGroupRequest(request)
      case ApiKeys.LIST_GROUPS => handleListGroupsRequest(request)
      case ApiKeys.SASL_HANDSHAKE => handleSaslHandshakeRequest(request)
      case ApiKeys.API_VERSIONS => handleApiVersionsRequest(request)
      case ApiKeys.CREATE_TOPICS => handleCreateTopicsRequest(request)
      case ApiKeys.DELETE_TOPICS => handleDeleteTopicsRequest(request)
      case ApiKeys.DELETE_RECORDS => handleDeleteRecordsRequest(request)
      case ApiKeys.INIT_PRODUCER_ID => handleInitProducerIdRequest(request)
      case ApiKeys.OFFSET_FOR_LEADER_EPOCH => handleOffsetForLeaderEpochRequest(request)
      case ApiKeys.ADD_PARTITIONS_TO_TXN => handleAddPartitionToTxnRequest(request)
      case ApiKeys.ADD_OFFSETS_TO_TXN => handleAddOffsetsToTxnRequest(request)
      case ApiKeys.END_TXN => handleEndTxnRequest(request)
      case ApiKeys.WRITE_TXN_MARKERS => handleWriteTxnMarkersRequest(request)
      case ApiKeys.TXN_OFFSET_COMMIT => handleTxnOffsetCommitRequest(request)
      case ApiKeys.DESCRIBE_ACLS => handleDescribeAcls(request)
      case ApiKeys.CREATE_ACLS => handleCreateAcls(request)
      case ApiKeys.DELETE_ACLS => handleDeleteAcls(request)
      case ApiKeys.ALTER_CONFIGS => handleAlterConfigsRequest(request)
      case ApiKeys.DESCRIBE_CONFIGS => handleDescribeConfigsRequest(request)
      case ApiKeys.ALTER_REPLICA_LOG_DIRS => handleAlterReplicaLogDirsRequest(request)
      case ApiKeys.DESCRIBE_LOG_DIRS => handleDescribeLogDirsRequest(request)
      case ApiKeys.SASL_AUTHENTICATE => handleSaslAuthenticateRequest(request)
      case ApiKeys.CREATE_PARTITIONS => handleCreatePartitionsRequest(request)
    }
  } catch {
    case e: FatalExitError => throw e
    case e: Throwable => handleError(request, e)
  } finally {
    request.apiLocalCompleteTimeNanos = time.nanoseconds
  }
}