深入理解Kafka服务端请求队列中请求的处理

    在《深入理解Kafka服务端之Processor线程是如何工作的》中&＃xff0c;通过分析得知Processor线程最终将接收到的客户端请求封装成Request对象&＃xff0c;放入了RequestChannel的requestQueue请求队列中。那么这些队列中的请求是如何被处理的呢&＃xff1f;这篇进行详细分析。

三、源码分析

    既然请求都被封装成Request对象放到了请求队列中&＃xff0c;那么就肯定会有一个线程去获取这些请求对象&＃xff0c;进行相应的处理。在之前Acceptor线程启动的过程中&＃xff0c;我们从服务端程序入口(即Kafka.main()方法)开始&＃xff0c;一直找到了KafkaServer.startup()方法&＃xff0c;当时提到过&＃xff0c;整个Kafka服务端的功能都在这个startup()方法中启动&＃xff0c;那么继续从startup方法中查找处理请求队列的相应代码&＃xff0c;结果如下&＃xff1a;

//TODO 创建处理Request请求的线程池&＃xff0c;这里的numIoThreads就是线程池的容量&＃xff0c;由服务端参数num.io.threads决定&＃xff0c;默认为8dataPlaneRequestHandlerPool &＃61; new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time, config.numIoThreads, s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent", SocketServer.DataPlaneThreadPrefix)前面分析Kafka网络架构采用了主从Reactor多线程模型时&＃xff0c;提到真正的网络I/O操作会交给I/O线程池中的I/O线程完成&＃xff0c;那么这里的KafkaRequestHandlerPool就是处理请求的I/O线程池。

    1. 查看的KafkaRequestHandlerPool主构造函数如下&＃xff1a;

class KafkaRequestHandlerPool(val brokerId: Int, val requestChannel: RequestChannel, val apis: KafkaApis, time: Time, numThreads: Int, requestHandlerAvgIdleMetricName: String, logAndThreadNamePrefix : String) extends Logging with KafkaMetricsGroup { //线程池中线程数量  private val threadPoolSize: AtomicInteger &＃61; new AtomicInteger(numThreads) private val aggregateIdleMeter &＃61; newMeter(requestHandlerAvgIdleMetricName, "percent", TimeUnit.NANOSECONDS) this.logIdent &＃61; "[" &＃43; logAndThreadNamePrefix &＃43; " Kafka Request Handler on Broker " &＃43; brokerId &＃43; "], " //管理线程的数组 val runnables &＃61; new mutable.ArrayBuffer[KafkaRequestHandler](numThreads) for (i 0 until numThreads) { //创建IO处理线程 createHandler(i) } ...这里重点看几个参数&＃xff1a;

• requestChannel&＃xff1a;SocketServer 中的请求通道对象。由于KafkaRequestHandlerPool是线程池对象&＃xff0c;那么内部线程处理的请求来源在哪儿&＃xff1f;请求恰恰是保存在 RequestChannel 中的请求队列requestQueue中&＃xff0c;因此&＃xff0c;Kafka 在构造 KafkaRequestHandlerPool实例时&＃xff0c;必须关联 SocketServer 组件中的 RequestChannel 实例&＃xff0c;让 I/O 线程能够找到请求被保存的地方。

• apis&＃xff1a;KafkaApis对象。IO线程会将拿到的请求对象Request交给KafkaApis去执行真正的逻辑处理。

• numThreads&＃xff1a;线程池中线程的数量。初始化线程池时要创建多少个线程由这个参数决定。而这个参数又是由broker端的参数num.io.threads决定的&＃xff0c;默认为8。

    除了主构造函数&＃xff0c;再来看几个KafkaRequestHandlerPool的方法&＃xff1a;

    a. createHandler(id:Int)&＃xff1a;创建线程的方法

• 创建KafkaRequestHandler线程对象

• 将线程对象放入runnables数组

• 设置线程为守护线程并启动

def createHandler(id: Int): Unit &＃61; synchronized { //创建KafkaRequestHandler线程&＃xff0c;并放入runnables数组 runnables &＃43;&＃61; new KafkaRequestHandler(id, brokerId, aggregateIdleMeter, threadPoolSize, requestChannel, apis, time) //设置线程为守护线程&＃xff0c;并启动 KafkaThread.daemon(logAndThreadNamePrefix &＃43; "-kafka-request-handler-" &＃43; id, runnables(id)).start()}

    b. resizeThreadPool(newSize: Int)&＃xff1a;根据给定值调整线程池的线程数量

• 如果给定的值大于当前线程池中的线程数&＃xff0c;则创建差值个新的线程&＃xff0c;放入线程池并启动

• 如果给定的值小于当前线程池中的线程数&＃xff0c;则将多余的线程从线程池中移除&＃xff0c;并关闭移除的线程

• 更新线程池中线程的数量

def resizeThreadPool(newSize: Int): Unit &＃61; synchronized { val currentSize &＃61; threadPoolSize.get info(s"Resizing request handler thread pool size from $currentSize to $newSize") //如果给定的值大于当前线程池的容量&＃xff0c;则创建差值个新的线程&＃xff0c;放入线程池并启动 if (newSize > currentSize) { for (i createHandler(i) } //如果给定的值小于当前线程池容量&＃xff0c;则将多余的线程从线程池中移除&＃xff0c;并关闭 } else if (newSize

    2. 这里我们再看一下KafkaRequestHandler的主构造函数&＃xff1a;

class KafkaRequestHandler(id: Int, //IO线程编号 brokerId: Int, val aggregateIdleMeter: Meter, val totalHandlerThreads: AtomicInteger,//线程池中线程数量 val requestChannel: RequestChannel,//请求处理的通道&＃xff0c;里面包含了requestQueue apis: KafkaApis,//KafkaApis类&＃xff0c;用于真正实现请求处理逻辑的类 time: Time) extends Runnable with Logging { this.logIdent &＃61; "[Kafka Request Handler " &＃43; id &＃43; " on Broker " &＃43; brokerId &＃43; "], "  //定义一个CountDownLatch对象&＃xff0c;便于管理线程的执行 private val shutdownComplete &＃61; new CountDownLatch(1) &＃64;volatile private var stopped &＃61; false ...参数apis、requestChannel和中KafkaRequestHandlerPool的一样&＃xff1a;

• id&＃xff1a;这里的id是线程池中的线程序号

• totalHandlerThreads&＃xff1a;线程池中的线程总数