当前位置：首页 > news >正文

Kafka-服务端-网络层-源码流程

news 来源：原创 2024/7/7 18:01:42

整体架构如下所示：

在这里插入图片描述

responseQueue不在RequestChannel中，在Processor中，每个Processor内部有一个responseQueue

客户端发送的请求被Acceptor转发给Processor处理
处理器将请求放到RequestChannel的requestQueue中
KafkaRequestHandler取出requestQueue中的请求
调用KafkaApis进行业务逻辑处理
KafkaApis将响应结果放到对应的Processor的responseQueue中
processor从responseQueue中取出响应结果
processor将响应结果返回给客户端

KafkaServer是Kafka服务端的主类，KafkaServer中和网络成相关的服务组件包括SocketServer、KafkaApis和KafkaRequestHandlerPool。SocketServer主要关注网络层的通信协议，具体的业务处理逻辑则交给KafkaRequestHandler和KafkaApis来完成。

class KafkaServer(val config: KafkaConfig) {def startup() {socketServer = new SocketServer(config, metrics, time, credentialProvider)socketServer.startup(startupProcessors = false)/* start processing requests */apis = new KafkaApis(socketServer.requestChannel, ...)requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, ...)}}

SocketServer

  def startup(startupProcessors: Boolean = true) {this.synchronized {...createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)if (startupProcessors) {startProcessors()}}private def createAcceptorAndProcessors(processorsPerListener: Int,endpoints: Seq[EndPoint]): Unit = synchronized {...endpoints.foreach { endpoint =>...val acceptor = new Acceptor(endpoint, ...)addProcessors(acceptor, endpoint, processorsPerListener)KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()acceptor.awaitStartup()acceptors.put(endpoint, acceptor)}}

可以看出SocketServer.startup()中会根据listener的个数创建相同个数的acceptor，每个acceptor关联数个processor。这是一种典型的Reactor模式，acceptor负责与客户端建立连接，并将连接分发给processor，processor负责所分连接后续的所有读写交互。

Acceptor

  def run() {serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)startupComplete()try {var currentProcessor = 0while (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.nextiter.remove()if (key.isAcceptable) {val processor = synchronized {currentProcessor = currentProcessor % processors.sizeprocessors(currentProcessor)}accept(key, processor)} elsethrow new IllegalStateException("Unrecognized key state for acceptor thread.")// round robin to the next processor thread, mod(numProcessors) will be done latercurrentProcessor = currentProcessor + 1} 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 ecase e: Throwable => error("Error occurred", e)}}} finally {debug("Closing server socket and selector.")CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)shutdownComplete()}}

上面是Acceptor的run()方法，可以看出，Acceptor在通道上注册了SelectionKey.OP_ACCEPT事件（OP_READ、OP_WRITE、OP_CONNECT、OP_ACCEPT，客户端监听OP_CONNECT事件，负责发起连接，服务端监听OP_CONNECT事件，负责建立连接），负责与客户端建立连接。并将建立的连接通过轮询的方式指派给processor。

Processor

每个Processor都会分到数个与客户端的连接。Processor的处理逻辑如下所示：

  override def run() {startupComplete()try {while (isRunning) {try {// 在新分到的客户端连接上注册OP_READ事件configureNewConnections()// 从responseQueue中取响应，赋值给KafkaChannel的send，等待poll时发送processNewResponses()// selector轮询各种事件，读取请求或者发送响应poll()// 封装selector.completedReceives中的请求，放入requestQueueprocessCompletedReceives()// 处理selector.completedSends响应（移除inflightResponses中的记录；执行响应的回调函数）processCompletedSends()processDisconnected()} catch {...}}} finally {...}}

Processor线程的名字中有kafka-network字样，可以通过jstack -l pid | grep kafka-network进行筛选。

KafkaRequestHandlerPool

KafkaServer会创建请求处理线程池（KafkaRequestHandlerPool），在请求处理线程池中会创建并启动多个请求处理线程（KafkaRequestHandler）。KafkaRequestHandler会获取RequestChannel.requestQueue中的请求进行处理，在内部实际处理会交给KafkaApis完成。

class KafkaRequestHandlerPool(val brokerId: Int, ...) {...for (i <- 0 until numThreads) {createHandler(i)}def createHandler(id: Int): Unit = synchronized {runnables += new KafkaRequestHandler(..., requestChannel, apis, time)KafkaThread.daemon("kafka-request-handler-" + id, runnables(id)).start()}
}

KafkaRequestHandler的run()方法如下：

class KafkaRequestHandler(id: Int,...) extends Runnable with Logging {...def run() {while (!stopped) {val req = requestChannel.receiveRequest(300)req match {case RequestChannel.ShutdownRequest =>shutdownComplete.countDown()returncase request: RequestChannel.Request =>try {request.requestDequeueTimeNanos = endTimeapis.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()}}