掘金小册-Kafka源码精讲20、21、22

avatar
Administrator

2024-10-22

41
0

20.服务端通信层：RequestChannel 类、KafkaRequestHandler 类和 KafkaApis 类

以下是对 RequestChannel 类、KafkaRequestHandler 类和 KafkaApis 类的介绍：


一、RequestChannel 类


  1. 功能概述

    • RequestChannel 在 Kafka 服务端通信层中扮演着请求和响应的传递管道的角色。它负责接收来自客户端的请求，并将处理后的响应返回给客户端。可以将其看作是连接客户端与服务端内部处理逻辑的桥梁。

  2. 主要组成部分

    • 请求队列：存储待处理的客户端请求。当客户端发送请求时，这些请求会被放入请求队列中，等待被服务端处理。例如，生产者发送消息的请求和消费者拉取消息的请求都会在这里排队等待处理。

    • 响应队列：用于存放处理完成后的响应。服务端处理完请求后，将响应放入这个队列，然后通过网络通信层将响应发送回客户端。

    • 控制信号：用于协调请求的处理和响应的发送。例如，可以通过控制信号来暂停或恢复请求的处理，或者通知其他组件有新的请求或响应可用。

  3. 重要性

    • 确保请求的有序处理。通过请求队列和响应队列的设计，保证了请求和响应的顺序性，避免了混乱和数据丢失。

    • 实现异步处理。客户端可以在发送请求后继续执行其他任务，而不必等待响应立即返回。服务端可以在后台处理请求，并在完成后将响应放入响应队列，提高了系统的并发性能。


二、KafkaRequestHandler 类


  1. 功能概述

    • KafkaRequestHandler 负责从 RequestChannel 中获取请求并进行实际的业务处理。它是连接 RequestChannel 和 KafkaApis 的中间层，将请求传递给具体的业务处理逻辑，并将处理结果返回给 RequestChannel。

  2. 处理流程

    • 获取请求：从 RequestChannel 的请求队列中获取待处理的请求。这可以通过轮询请求队列或者使用事件驱动的方式实现，确保及时处理新到达的请求。

    • 调用业务逻辑：根据请求的类型，调用相应的 KafkaApis 中的方法进行业务处理。例如，如果是生产者发送消息的请求，会调用 KafkaApis 中与消息存储相关的方法；如果是消费者拉取消息的请求，会调用与消息检索相关的方法。

    • 生成响应：根据业务处理的结果，生成相应的响应对象。这个响应对象包含了请求的处理状态（成功或失败）以及相应的数据（如消息内容、错误信息等）。

    • 返回响应：将生成的响应对象放入 RequestChannel 的响应队列中，以便后续发送给客户端。

  3. 重要性

    • 实现业务逻辑的隔离。将业务处理逻辑与网络通信层分离，使得代码结构更加清晰，易于维护和扩展。

    • 提高处理效率。可以通过多线程的方式同时处理多个请求，充分利用服务器的资源，提高系统的吞吐量。


三、KafkaApis 类


  1. 功能概述

    • KafkaApis 是 Kafka 服务端的核心业务逻辑实现类。它包含了处理各种类型请求的具体方法，如处理生产者发送消息、消费者拉取消息、创建主题、删除主题等请求。

  2. 主要方法

    • 消息存储相关方法：用于处理生产者发送消息的请求。包括确定消息的目标分区、将消息写入分区存储、管理副本同步等操作。

    • 消息检索相关方法：用于处理消费者拉取消息的请求。包括根据消费者的请求参数定位消息位置、读取消息并返回给消费者、更新消费者的偏移量等操作。

    • 主题管理相关方法：用于处理创建主题、删除主题、修改主题配置等请求。包括在元数据中创建主题的记录、分配分区到不同的 broker、更新主题的配置信息等操作。

  3. 重要性

    • 实现 Kafka 的核心业务功能。KafkaApis 中的方法直接实现了 Kafka 作为分布式消息系统的主要功能，是整个系统的核心所在。

    • 提供统一的业务接口。其他组件可以通过调用 KafkaApis 中的方法来实现各种业务操作，无需了解具体的实现细节，提高了系统的可维护性和可扩展性。

在 Kafka 的服务端通信层中，RequestChannel 类、KafkaRequestHandler 类和 KafkaApis 类是核心组件，它们共同协作来处理客户端的请求和响应。下面分别介绍这些类的作用和它们之间的交互：

1. RequestChannel 类

RequestChannel 是一个核心的类，它负责管理网络请求的接收和响应的发送。它维护了以下几个关键的组件：

  • 请求队列（Request Queue）：RequestChannel 包含一个请求队列，用于存储从网络层接收到的请求，等待被处理。

  • 响应队列（Response Queue）：每个 Processor 线程都有一个与之关联的响应队列，用于存储处理完的响应，等待被发送回客户端。

RequestChannel 的主要方法包括：

  • enqueueResponse：将处理完成的响应添加到响应队列中。

  • dequeueRequest：从请求队列中取出请求进行处理。

2. KafkaRequestHandler 类

KafkaRequestHandler 是工作线程，它负责从 RequestChannel 的请求队列中取出请求，进行处理，并生成响应。这些线程通常位于一个线程池 KafkaRequestHandlerPool 中，线程池的大小由配置参数 num.io.threads 决定。

KafkaRequestHandler 的主要工作流程：

  • 请求处理：从请求队列中取出请求，根据请求的类型调用相应的 KafkaApis 方法进行处理。

  • 生成响应：处理完请求后，生成响应数据，并将其通过 RequestChannel 返回给对应的 Processor 线程。

  • 异常处理：如果请求处理过程中出现异常，KafkaRequestHandler 也会负责生成错误响应。

3. KafkaApis 类

KafkaApis 是一个接口类，它定义了 Kafka 能够处理的各种请求类型（如生产者请求、消费者请求、元数据请求等）的处理逻辑。具体的请求处理逻辑由 KafkaApis 的实现类提供。

KafkaApis 的主要方法包括：

  • handleProducerRequest：处理生产者请求。

  • handleConsumerRequest：处理消费者请求。

  • handleMetadataRequest：处理元数据请求。

组件间的交互

  1. 网络层到 RequestChannel：当网络层（Processor 线程）从客户端接收到请求时，它会将请求封装成 RequestChannel.Request 对象，并将其放入 RequestChannel 的请求队列中。

  2. RequestChannel 到 KafkaRequestHandler：KafkaRequestHandler 线程从 RequestChannel 的请求队列中取出请求，然后根据请求类型调用 KafkaApis 中相应的处理方法。

  3. KafkaRequestHandler 到 RequestChannel：处理完请求后，KafkaRequestHandler 将生成的响应通过 RequestChannel 返回给对应的 Processor 线程，由 Processor 线程负责将响应发送回客户端。

通过这种设计，Kafka 能够将网络通信、请求处理和业务逻辑分离，提高了系统的可维护性和扩展性。同时，这种多线程的处理模式也使得 Kafka 能够有效地处理高并发的请求。

上节课我们学习了Acceptor和Processor这两个类，这两个负责的是网络操作并不涉及业务执行，比如消息的存储等，那么业务执行是由哪个组件执行的呢？其实是通过KafkaRequestHandler线程类的线程池完成的，但是KafkaRequestHandler并没有直接与Processor类接触，而是通过RequestChannel类作为一个桥梁沟通的。

好，我们今天主要介绍RequestChannel类和KafkaRequestHandler线程类。

RequestChannel 类

同样，我们先了解一下这个类的重要的字段。以下是源码以及相应的注释。

重要字段

class RequestChannel(val queueSize: Int, val metricNamePrefix : String,
time: Time) extends KafkaMetricsGroup {
import RequestChannel._   
val metrics = new RequestChannel.Metrics   
//请求队列,queueSize为队列大小。默认500   
private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
//RequestChannel下辖的Processor线程池。默认8个线程。   
private val processors = new ConcurrentHashMap[Int, Processor]()

  • requestQueue：元素为Request对象的阻塞队列。队列长度大小为500。这个队列的作用是Processor类接收到请求后会把请求放入这个队列里。requestQueue里的元素的消费方是KafkaRequestHandler线程类。

  • processors ：RequestChannel下辖的Processor线程池。默认8个线程，Kafka 允许你动态地修改此参数值。Broker 启动时指定 num.network.threads 为 8，之后你通过 kafka-configs 命令修改。

好，接下来我们学习重要的方法。

重要方法

首先，我们学习一下RequestChannel类怎么把Processor类对象添加到Processors集合的。

方法 addProcessor(）

def addProcessor(processor: Processor): Unit = {     
// 添加Processor到Processor线程池     
if (processors.putIfAbsent(processor.id, processor) != null)
warn(s"Unexpected processor with processorId ${processor.id}")     
// 为给定Processor对象创建对应的监控指标
newGauge(responseQueueSizeMetricName, 
() => processor.responseQueueSize,       
Map(ProcessorMetricTag -> processor.id.toString)) 
}

这个方法很简单就是把processor放入processors集合里，只是新放入之前会判断processor.id是否已经有了。

方法 sendRequest()

这个方法很重要，是用来暂时存储请求的。

def sendRequest(request: RequestChannel.Request): Unit = {     requestQueue.put(request) }

Porcessor类对象接收到请求后，会调用这个方法把请求放入requestQueue阻塞队列里。等待KafkaRequestHandler线程类获取请求。

方法 receiveRequest()

表示从requestQueue队列中取出请求。

  def receiveRequest(timeout: Long): RequestChannel.BaseRequest =     requestQueue.poll(timeout, TimeUnit.MILLISECONDS)

KafkaRequestHandler线程类会调用这个方法取出request对象，再根据request对象调用底层对应的api。