Kafka请求队列源码实现-RequestChannel请求通道_kafka的memorypool类

Kafka如何实现请求队列

• 核心类
  

Kafka服务端，即Broker，负责消息的持久化，是个不断接收外部请求、处理请求，然后发送处理结果的Java进程。

Broker的高处理性能在于高效保存排队中的请求。
Broker底层请求对象的建模
请求队列的实现原理
Broker请求处理方面的核心监控指标。

Broker与Clients主要基于Request/Response机制交互，所以看看如何建模或定义Request和Response。

请求（Request）

定义了Kafka Broker支持的各类请求。

• RequestChannel#Request
  
  
  trait关键字类似于Java的interface。从代码中，我们可以知道，
  ShutdownRequest只做标志位。当Broker进程关闭时，RequestHandler会发送ShutdownRequest到专属请求处理线程。该线程接收到此请求后，会主动触发Broker关闭流程。

Request才是真正的定义各类Clients端或Broker端请求的实现类。

属性

processor

Processor线程的序号，即该请求由哪个Processor线程接收处理。

• Broker端参数num.network.threads控制Broker每个监听器上创建的Processor线程数
  

假设listeners配置为PLAINTEXT://localhost:9092,SSL://localhost:9093，则默认情况下Broker启动时会创建6个Processor线程，每3个为一组，分别给listeners参数中设置的两个监听器使用，每组的序号分别是0、1、2。

为什么保存Processor线程序号？

当Request被后面的I/O线程处理完成后，还要依靠Processor线程发送Response给请求方，因此，Request必须记录它之前被哪个Processor线程接收。
Processor线程只是网络接收线程，并不会执行真正的I/O线程才负责的Request请求处理逻辑。

context

• 用于标识请求上下文信息，RequestContext类维护Request的所有上下文信息。
  

RequestContext类

startTimeNanos

维护Request对象被创建的时间，用于计算各种时间统计指标。

请求对象中的很多JMX（Java Management Extensions）指标，特别是时间类统计指标，都需要startTimeNanos字段，纳秒单位的时间戳信息，可实现细粒度时间统计精度。

memoryPool

一个非阻塞式内存缓冲区，用于避免Request对象无限使用内存。

内存缓冲区的接口类MemoryPool，实现类SimpleMemoryPool。可重点关注下SimpleMemoryPool#tryAllocate，怎么为Request对象分配内存。

    @Override
    public ByteBuffer tryAllocate(int sizeBytes) {
        if (sizeBytes < 1)
            throw new IllegalArgumentException("requested size " + sizeBytes + "<=0");
        if (sizeBytes > maxSingleAllocationSize)
            throw new IllegalArgumentException("requested size " + sizeBytes + " is larger than maxSingleAllocationSize " + maxSingleAllocationSize);

        long available;
        boolean success = false;
        //in strict mode we will only allocate memory if we have at least the size required.
        //in non-strict mode we will allocate memory if we have _any_ memory available (so available memory
        //can dip into the negative and max allocated memory would be sizeBytes + maxSingleAllocationSize)
        long threshold = strict ? sizeBytes : 1;
        while ((available = availableMemory.get()) >= threshold) {
            success = availableMemory.compareAndSet(available, available - sizeBytes);
            if (success)
                break;
        }

        if (success) {
            maybeRecordEndOfDrySpell();
        } else {
            if (oomTimeSensor != null) {
                startOfNoMemPeriod.compareAndSet(0, System.nanoTime());
            }
            log.trace("refused to allocate buffer of size {}", sizeBytes);
            return null;
        }

        ByteBuffer allocated = ByteBuffer.allocate(sizeBytes);
        bufferToBeReturned(allocated);
        return allocated;
    }
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

buffer

真正保存Request对象内容的字节缓冲区。Request发送方须按Kafka RPC协议规定格式向该缓冲区写入字节，否则抛InvalidRequestException。

• 该逻辑由RequestContext#parseRequest实现。

ApiVersions请求作用

当Broker接收到ApiVersionsRequest，它会返回Broker当前支持的请求类型列表，包括请求类型名称、支持的最早版本号和最新版本号。查看Kafka的bin目录，能找到kafka-broker-api-versions.sh脚本工具。它就是，构造ApiVersionsRequest对象，然后发送给对应的Broker。

若是ApiVersions类型请求，代码中为什么要判断一下它的版本呢？
和处理其他类型请求不同，Kafka必须保证版本号比最新支持版本还要高的ApiVersions请求也能被处理。这主要是考虑客户端和服务器端版本兼容。客户端发请求给Broker，可能不知道Broker到底支持哪些版本请求，它需使用ApiVersionsRequest去获取完整请求版本支持列表。若不做该判断，Broker可能无法处理客户端发送的ApiVersionsRequest。

metrics

metrics是Request相关的各种监控指标的一个管理类。它构建了一个Map，封装了所有请求JMX指标。

响应（Response）

定义了与Request对应的各类响应。

类设计

• Response
  定义Response的抽象父类。每个Response对象都包含对应Request对象。该类核心方法onComplete，用来实现每类Response被处理后需要执行的回调逻辑。
• SendResponse
  大多数Request处理完成后都需执行一段回调，SendResponse即保存返回结果的Response子类。核心字段onCompletionCallback，即指定处理完成之后的回调逻辑。

正常需要发送Response。

• NoResponse
  有些Request处理完成后无需单独执行额外的回调逻辑。NoResponse就是为这类Response准备的。

无需发送Response。

• CloseConnectionResponse
  出错后需要关闭TCP连接的场景，此时返回CloseConnectionResponse给Request发送方，显式地通知它关闭连接。

标识关闭连接通道的Response。

• StartThrottlingResponse
  通知Broker的Socket Server组件（后面几节课我会讲到它）某个TCP连接通信通道开始被限流（throttling）。
• EndThrottlingResponse
  与StartThrottlingResponse对应，通知Broker的SocketServer组件某个TCP连接通信通道的限流已结束。

后两个Response类不常用，仅在对Socket连接进行限流时，才会使用。

Response代码

abstract class Response(val request: Request) {
  locally {
    val nowNs = Time.SYSTEM.nanoseconds
    request.responseCompleteTimeNanos = nowNs
    if (request.apiLocalCompleteTimeNanos == -1L)
      request.apiLocalCompleteTimeNanos = nowNs
  }
  def processor: Int = request.processor
  def responseString: Option[String] = Some("")
  def onComplete: Option[Send => Unit] = None
  override def toString: String
}
1
2
3
4
5
6
7
8
9
10
11
12

该抽象类只有一个属性字段：request。即每个Response对象都要保存它对应的Request对象。
onComplete方法是调用指定回调逻辑的地方。

SendResponse类就重写了该方法：

class SendResponse(request: Request,
                     val responseSend: Send,
                     val responseAsString: Option[String],
                     val onCompleteCallback: Option[Send => Unit]) 
  extends Response(request) {
    ......
    // 指定输入参数onCompleteCallback
    override def onComplete: Option[Send => Unit] = onCompleteCallback
}
1
2
3
4
5
6
7
8
9

onComplete方法把函数赋值给另一个函数，并作为结果返回。好处在于可以灵活变更onCompleteCallback实现不同回调逻辑。

RequestChannel

实现了Kafka Request队列。传输Request/Response的通道。有了Request和Response的基础，下面我们可以学习RequestChannel类的实现了。

定义

RequestChannel类实现KafkaMetricsGroup trait，后者封装许多实用指标监控方法：

• newGauge
  创建数值型监控指标
• newHistogram
  创建直方图型监控指标

属性

每个RequestChannel对象实例创建时，会定义队列保存Broker接收到的各类请求，这个队列被称为请求队列或Request队列。
Kafka使用Java提供的阻塞队列ArrayBlockingQueue实现请求队列，并利用它天然提供的线程安全保证多个线程能够并发安全高效地访问请求队列。

• 代码中该队列由变量requestQueue定义。
  

queueSize

Request队列的最大长度。当Broker启动时，SocketServer组件会创建RequestChannel对象

并把Broker端参数queued.max.requests赋值给queueSize。默认情况每个RequestChannel上的队列长度500。

processors

封装RequestChannel的Processor线程池。每个Processor线程负责具体的请求处理逻辑。

Processor管理

刚才的processors即是被创建的Processor线程池，使用Java#ConcurrentHashMap保存：

• Key：processor序号
• Value：具体的Processor线程对象

因此当前Kafka Broker端所有网络线程都是在RequestChannel中维护的。

管理线程池

• RequestChannel中的addProcessor和removeProcessor。
  
  

分别实现增加和移除线程。每当Broker启动，都会调用addProcessor方法，向RequestChannel对象添加num.network.threads个Processor线程。

num.network.threads这个参数的更新模式（Update Mode）是Cluster-wide，即Kafka允许你动态修改此参数值。比如，Broker启动时指定num.network.threads为8，之后你通过kafka-configs命令将其修改为3。显然该操作会减少Processor线程池中的线程数量。在这个场景下，removeProcessor方法会被调用。

处理Request和Response

即收发Request和发送Response。
sendRequest和receiveRequest：

• 发送Request仅是将Request对象置于Request队列
  
• 接收Request则是从队列中取出Request

整个流程其实就是“生产者-消费者”模式，依靠ArrayBlockingQueue的线程安全确保整个过程的线程安全

没有所谓的接收Response，只有发送Response，即sendResponse方法。sendResponse是啥意思呢？其实就是把Response对象发送出去，也就是将Response添加到Response队列的过程。

当Processor处理完某个Request后，会把自己的序号封装进对应的Response对象。
一旦找出之前是由哪个Processor线程处理，代码直接调用该Processor的enqueueResponse方法，将Response放入Response队列中，等待后续发送。

监控指标

RequestChannel类定义封装了与Request队列相关的重要监控指标，以实时动态地监测Request和Response的性能表现。

具体指标项

object RequestMetrics {
  val consumerFetchMetricName = ApiKeys.FETCH.name + "Consumer"
  val followFetchMetricName = ApiKeys.FETCH.name + "Follower"
  val RequestsPerSec = "RequestsPerSec"
  val RequestQueueTimeMs = "RequestQueueTimeMs"
  val LocalTimeMs = "LocalTimeMs"
  val RemoteTimeMs = "RemoteTimeMs"
  val ThrottleTimeMs = "ThrottleTimeMs"
  val ResponseQueueTimeMs = "ResponseQueueTimeMs"
  val ResponseSendTimeMs = "ResponseSendTimeMs"
  val TotalTimeMs = "TotalTimeMs"
  val RequestBytes = "RequestBytes"
  val MessageConversionsTimeMs = "MessageConversionsTimeMs"
  val TemporaryMemoryBytes = "TemporaryMemoryBytes"
  val ErrorsPerSec = "ErrorsPerSec"
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• RequestsPerSec
  每秒处理的Request数，用来评估Broker的繁忙状态。
• RequestQueueTimeMs
  计算Request在Request队列中的平均等候时间，单位是毫秒。倘若Request在队列的等待时间过长，你通常需要增加后端I/O线程的数量，来加快队列中Request的拿取速度。
• LocalTimeMs
  计算Request实际被处理的时间，单位是毫秒。一旦定位到这个监控项的值很大，你就需要进一步研究Request被处理的逻辑了，具体分析到底是哪一步消耗了过多的时间。
• RemoteTimeMs
  Kafka的读写请求（PRODUCE请求和FETCH请求）逻辑涉及等待其他Broker操作的步骤。RemoteTimeMs计算的，就是等待其他Broker完成指定逻辑的时间。因为等待的是其他Broker，因此被称为Remote Time。这个监控项非常重要！Kafka生产环境中设置acks=all的Producer程序发送消息延时高的主要原因，往往就是Remote Time高。因此，如果你也碰到了这样的问题，不妨先定位一下Remote Time是不是瓶颈。
• TotalTimeMs
  计算Request被处理的完整流程时间。这是最实用的监控指标，没有之一！毕竟，我们通常都是根据TotalTimeMs来判断系统是否出现问题的。一旦发现了问题，我们才会利用前面的几个监控项进一步定位问题的原因。
  RequestChannel定义了updateMetrics方法，用于实现监控项的更新