文章开头依然放上 Kafka 的网络通信架构图。

第三节中阅读了 SocketServer 部分的源码，知道了请求时如何接收并返回的。这一节我们一起阅读一下 KafkaRequestHandler 的源码，看看 IO 线程是如何工作的。

IO线程池：KafkaRequestHandlerPool

KafkaRequestHandler 是 IO 线程，那么自然而然的，就应该有一个线程池去调度它，这就是 KafkaRequestHandlerPool。

我们可以在 KafkaServer 的 startup 方法中，可以看到创建这个线程池的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// KafkaApis 是 kafka 真正处理请求逻辑的地方
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
  kafkaController, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
  fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache)

// 创建 RequestHandlerPool 线程池
dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
  config.numIoThreads, s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.DataPlaneThreadPrefix)

// 如果启用了 control plane 特性
// 这里同样也会去创建一套 RequestProcessor 和 RequestHandlerPool
socketServer.controlPlaneRequestChannelOpt.foreach { controlPlaneRequestChannel =>
  controlPlaneRequestProcessor = new KafkaApis(controlPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
    kafkaController, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
    fetchManager, brokerTopicStats, clusterId, time, tokenManager, brokerFeatures, featureCache)

  controlPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor, time,
    1, s"${SocketServer.ControlPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.ControlPlaneThreadPrefix)
}

下面是 KafkaRequestHandlerPool 的源码，我们可以清楚的看到，在它构造方法中，通过 createHandler 方法创建了 numThreads 个 handler 线程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class KafkaRequestHandlerPool(val brokerId: Int,
                              val requestChannel: RequestChannel,
                              val apis: ApiRequestHandler,
                              time: Time,
                              numThreads: Int,
                              requestHandlerAvgIdleMetricName: String,
                              logAndThreadNamePrefix : String) extends Logging with KafkaMetricsGroup {

  // 线程池大小，由配置文件中的 num.io.threads 控制
  // 这个参数是支持动态调整的，因此是一个 AtomicInteger 类
  private val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)

  /* a meter to track the average free capacity of the request handlers */
  private val aggregateIdleMeter = newMeter(requestHandlerAvgIdleMetricName, "percent", TimeUnit.NANOSECONDS)

  this.logIdent = "[" + logAndThreadNamePrefix + " Kafka Request Handler on Broker " + brokerId + "], "

  // 负责保存子线程，即 KafkaRequestHandler 线程
  val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)

  // 创建 handler 线程
  for (i <- 0 until numThreads) {
    createHandler(i)
  }

  def createHandler(id: Int): Unit = synchronized {
    runnables += new KafkaRequestHandler(id, brokerId, aggregateIdleMeter, threadPoolSize, requestChannel, apis, time)
    KafkaThread.daemon(logAndThreadNamePrefix + "-kafka-request-handler-" + id, runnables(id)).start()
  }

  def resizeThreadPool(newSize: Int): Unit = synchronized {
    val currentSize = threadPoolSize.get
    info(s"Resizing request handler thread pool size from $currentSize to $newSize")
    if (newSize > currentSize) {
      for (i <- currentSize until newSize) {
        createHandler(i)
      }
    } else if (newSize < currentSize) {
      for (i <- 1 to (currentSize - newSize)) {
        runnables.remove(currentSize - i).stop()
      }
    }
    threadPoolSize.set(newSize)
  }

  def shutdown(): Unit = synchronized {
    info("shutting down")
    for (handler <- runnables) {
      // 向 RequestChannel.requestQueue() 中塞一个 ShutdownRequest
      handler.initiateShutdown()
    }
    for (handler <- runnables) {
      // 这里会阻塞， 直到 KafkaRequestHandler 读到 ShutdownRequest 的时候，
      // 并 shutdown 自己为止
      handler.awaitShutdown()
    }
    info("shut down completely")
  }
}

IO线程：KafkaRequestHandler

对于处理业务的 handler 线程，我们自然要看一看它的 run 方法了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def run(): Unit = {
  while (!stopped) {
    // We use a single meter for aggregate idle percentage for the thread pool.
    // Since meter is calculated as total_recorded_value / time_window and
    // time_window is independent of the number of threads, each recorded idle
    // time should be discounted by # threads.
    //
    // 跟踪线程池的总空闲时间比例
    // Meter 的值是通过 total_recorded_value / time_window 计算的
    // 其中:
    // total_recorded_value 是总的记录值。
    // time_window 是一个时间窗口，独立于线程数量。
    //
    // 因为这个指标是测量线程池中所有线程的整体空闲情况，而不是单个线程的。
    // 所以要除以线程数，对每个线程的空闲时间进行折算，照线程总数进行平摊。
    // 这样可以提供一个更准确的度量，显示线程池的整体闲置率，而不是单个线程的状态。
    val startSelectTime = time.nanoseconds

    // 其实就是从 RequestChannel 的 requestQueue 中取出一个请求
    val req = requestChannel.receiveRequest(300)

    val endTime = time.nanoseconds
    val idleTime = endTime - startSelectTime

    // 当前线程的空闲时间/线程数 进行折算
    aggregateIdleMeter.mark(idleTime / totalHandlerThreads.get)

    req match {

      // 如果是 shutdown 请求，则退出
      case RequestChannel.ShutdownRequest =>
        debug(s"Kafka request handler $id on broker $brokerId received shut down command")
        shutdownComplete.countDown()
        return

      // 正常的 request
      case request: RequestChannel.Request =>
        try {
          request.requestDequeueTimeNanos = endTime
          trace(s"Kafka request handler $id on broker $brokerId handling request $request")

          // 调用 KafkaApi 类去处理业务逻辑
          apis.handle(request)

        } catch {
          case e: FatalExitError =>
            shutdownComplete.countDown()
            Exit.exit(e.statusCode)
          case e: Throwable => error("Exception when handling request", e)
        } finally {
          request.releaseBuffer()
        }

      case null => // continue
    }
  }
  shutdownComplete.countDown()
}

到这里我们就和第三节连起来了，Processor 线程接收请求，将其封装为 Request，放到 RequestChannel 中的 requestQueue 中，然后 KafkaRequestHandler 线程循环去从这个队列中拉取请求，再通过 KafkaApis 类去实际处理。

KafkaApis: 处理业务

KafkaApis 是 Broker 端所有功能的入口，其中的 handle 方法封装了所有 RPC 请求的具体处理逻辑。

参考下面的代码，如果我们要自定义方法，只需要新增一个 ApiKeys 枚举，然后在这里新增 handleXXXRequest 方法就可以了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
override def handle(request: RequestChannel.Request): Unit = {
  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)
      case ApiKeys.CREATE_DELEGATION_TOKEN => handleCreateTokenRequest(request)
      case ApiKeys.RENEW_DELEGATION_TOKEN => handleRenewTokenRequest(request)
      case ApiKeys.EXPIRE_DELEGATION_TOKEN => handleExpireTokenRequest(request)
      case ApiKeys.DESCRIBE_DELEGATION_TOKEN => handleDescribeTokensRequest(request)
      case ApiKeys.DELETE_GROUPS => handleDeleteGroupsRequest(request)
      case ApiKeys.ELECT_LEADERS => handleElectReplicaLeader(request)
      case ApiKeys.INCREMENTAL_ALTER_CONFIGS => handleIncrementalAlterConfigsRequest(request)
      case ApiKeys.ALTER_PARTITION_REASSIGNMENTS => handleAlterPartitionReassignmentsRequest(request)
      case ApiKeys.LIST_PARTITION_REASSIGNMENTS => handleListPartitionReassignmentsRequest(request)
      case ApiKeys.OFFSET_DELETE => handleOffsetDeleteRequest(request)
      case ApiKeys.DESCRIBE_CLIENT_QUOTAS => handleDescribeClientQuotasRequest(request)
      case ApiKeys.ALTER_CLIENT_QUOTAS => handleAlterClientQuotasRequest(request)
      case ApiKeys.DESCRIBE_USER_SCRAM_CREDENTIALS => handleDescribeUserScramCredentialsRequest(request)
      case ApiKeys.ALTER_USER_SCRAM_CREDENTIALS => handleAlterUserScramCredentialsRequest(request)
      case ApiKeys.ALTER_ISR => handleAlterIsrRequest(request)
      case ApiKeys.UPDATE_FEATURES => handleUpdateFeatures(request)
      // Until we are ready to integrate the Raft layer, these APIs are treated as
      // unexpected and we just close the connection.
      case ApiKeys.VOTE => closeConnection(request, util.Collections.emptyMap())
      case ApiKeys.BEGIN_QUORUM_EPOCH => closeConnection(request, util.Collections.emptyMap())
      case ApiKeys.END_QUORUM_EPOCH => closeConnection(request, util.Collections.emptyMap())
      case ApiKeys.DESCRIBE_QUORUM => closeConnection(request, util.Collections.emptyMap())
    }
  } catch {
    case e: FatalExitError => throw e
    case e: Throwable => handleError(request, e)
  } finally {
    // try to complete delayed action. In order to avoid conflicting locking, the actions to complete delayed requests
    // are kept in a queue. We add the logic to check the ReplicaManager queue at the end of KafkaApis.handle() and the
    // expiration thread for certain delayed operations (e.g. DelayedJoin)
    replicaManager.tryCompleteActions()
    // The local completion time may be set while processing the request. Only record it if it's unset.
    if (request.apiLocalCompleteTimeNanos < 0)
      request.apiLocalCompleteTimeNanos = time.nanoseconds
  }
}

资源访问控制: authorizer

我会选择其中的两个方法详细学习，在此之前，我们需要先了解一下 authorize 方法。如果配置文件里没有配置 authorizer.class.name 参数，就不会启用身份验证，所有的操作都是开放的。如果配置了这个参数，在 KafkaApis 处理实际业务之前，都会用 authorize 方法判断一下 Request 的权限是否正确，比如说，这是一个创建 Topic 的请求，那么就要校验客户端有没有创建 Topic 的权限，成功之后才会继续执行，否则就会返回错误信息给客户端。

我们同样可以在 KafkaServer 的 startup() 方法中找到 authorizer 的创建和配置过程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
// 从配置文件的 authorizer.class.name 配置中读出应用的 auth 类，实例化并返回
authorizer = config.authorizer

// 调用 authorize 类的 configure 方法做具体的配置
authorizer.foreach(_.configure(config.originals))

// 开始加载 metadata，并返回一个可以表示 authorizer 已经异步初始化经完成的 CompletableFuture
//
// 目的是：用户自定义的 authorize 类在初始化过程中，可能需要一些异步的操作，为了保证
//        在 Acceptor 和 Processor开始接受请求的时候，这些异步操作全部都可以完成，
//        需要这么一个标志位
//
// 事实上：目前 kafka 所提供的 authorize 类中，并没有异步操作，这里返回的就是 completed Future
//         这是给用户自定义 authorize 类留下了空间
val authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = authorizer match {

  case Some(authZ) =>
    authZ.start(brokerInfo.broker.toServerInfo(clusterId, config)).asScala.map { case (ep, cs) =>
      ep -> cs.toCompletableFuture
    }

  case None =>
    // 如果没有配置 authorize 类，则什么都不做，返回一个 completed Future
    brokerInfo.broker.endPoints.map { ep =>
      ep.toJava -> CompletableFuture.completedFuture[Void](null)
    }.toMap
}

......

// 在 KafkaServer.startup() 方法的最后，启动 Acceptor 和 Processor 之前，
// 会调用 CompletableFuture.join()，等待上面的 CompletableFuture 完成
// 即：authorizer 真正初始化完成之后，kafka 才会去处理客户端请求
socketServer.startProcessingRequests(authorizerFutures)

具体 authorize 过程我们这里暂时不做研究。

发送响应 sendResponse

业务处理的细节目前暂不研究，我们看一下处理完业务之后，是如何发送响应的，下面是 KafkaApis 中的 sendResponse() 方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
private def sendResponse(request: RequestChannel.Request,
                         responseOpt: Option[AbstractResponse],
                         onComplete: Option[Send => Unit]): Unit = {
  // Update error metrics for each error code in the response including Errors.NONE
  responseOpt.foreach(response => requestChannel.updateErrorMetrics(request.header.apiKey, response.errorCounts.asScala))

  val response = responseOpt match {
    case Some(response) =>
      val responseSend = request.context.buildResponse(response)
      val responseString =
        if (RequestChannel.isRequestLoggingEnabled) Some(response.toString(request.context.apiVersion))
        else None
      new RequestChannel.SendResponse(request, responseSend, responseString, onComplete)
    case None =>
      new RequestChannel.NoOpResponse(request)
  }

  // 这里其实就是把响应放到了 RequestChannel 的 responseQueue 中
  // 等待上一节提到的 Processor 线程的 processNewResponses() 方法去处理
  requestChannel.sendResponse(response)
}

参考

Kafka 核心技术与实战 - 极客时间