Kafka 顺序消费线程模型的实践与优化

系统管理员 2024-04-08 10:51 58阅读 0赞

各类消息中间件对顺序消息实现的做法是将具有顺序性的一类消息发往相同的主题分区中，只需要将这类消息设置相同的 Key 即可，而 Kafka 会在任意时刻保证一个消费组同时只能有一个消费者监听消费，因此可在消费时按分区进行顺序消费，保证每个分区的消息具备局部顺序性。由于需要确保分区消息的顺序性，并不能并发地消费消息，对消费的吞吐量会造成一定的影响。那么，如何在保证消息顺序性的前提下，最大限度的提高消费者的消费能力？

**本文将会对 Kafka 消费者拉取消息流程进行深度分析之后，对 Kafka 消费者顺序消费线程模型进行一次实践与优化**。

### Kafka 消费者拉取消息流程分析 ###

在讲实现 Kafka 顺序消费线程模型之前，我们需要先深入分析 Kafka 消费者的消息拉取机制，只有当你对 Kafka 消费者拉取消息的整个流程有深入的了解之后，你才能够很好地理解本次线程模型改造的方案。

我先给大家模拟一下消息拉取的实际现象，这里 max.poll.records = 500。

1.  消息没有堆积时：

![5edeff5b69374ebaa86623fd56e68d1f.png][]

可以发现，在消息没有堆积时，消费者拉取时，如果某个分区没有的消息不足 500 条，会从其他分区凑够 500 条后再返回。

1.  多个分区都有堆积时：

![8f21ef090da24e849d4ba0348a56b5bc.png][]

在消息有堆积时，可以发现每次返回的都是同一个分区的消息，但经过不断 debug，消费者在拉取过程中并不是等某个分区消费完没有堆积了，再拉取下一个分区的消息，而是不断循环的拉取各个分区的消息，但是这个循环并不是说分区 p0 拉取完 500 条，后面一定会拉取分区 p1 的消息，很有可能后面还会拉取 p0 分区的消息，为了弄明白这种现象，我仔细阅读了相关源码。

org.apache.kafka.clients.consumer.KafkaConsumer\#poll

*  private ConsumerRecords<K, V> poll(final Timer timer, final boolean includeMetadataInTimeout) \{
 *  try \{
 *  // poll for new data until the timeout expires
 *  do \{
 *  // 客户端拉取消息核心逻辑
 *  final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(timer);
 *  if (!records.isEmpty()) \{
 *  // 在返回数据之前, 发送下次的 fetch 请求, 避免用户在下次获取数据时线程阻塞
 *  if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) \{
 *  // 调用 ConsumerNetworkClient\#poll 方法将 FetchRequest 发送出去。
 *  client.pollNoWakeup();
 *  \}
 *  return this.interceptors.onConsume(new ConsumerRecords<>(records));
 *  \}
 *  \} while (timer.notExpired());
 *  return ConsumerRecords.empty();
 *  \} finally \{
 *  release();
 *  \}
 *  \}
 *

我们使用 Kafka consumer 进行消费的时候通常会给一个时间，比如：

*  consumer.poll(Duration.ofMillis(3000));
 *

从以上代码逻辑可以看出来，用户给定的这个时间，目的是为了等待消息凑够 max.poll.records 条消息后再返回，即使消息条数不够 max.poll.records 消息，时间到了用户给定的等待时间后，也会返回。

pollForFetches 方法是客户端拉取消息核心逻辑，但并不是真正去 broker 中拉取，而是从缓存中去获取消息。在 pollForFetches 拉取消息后，如果消息不为零，还会调用 fetcher.sendFetches() 与 client.pollNoWakeup()，调用这两个方法究竟有什么用呢？

fetcher.sendFetches() 经过源码阅读后，得知该方法目的是为了构建拉取请求 FetchRequest 并进行发送，但是这里的发送并不是真正的发送，而是将 FetchRequest 请求对象存放在 unsend 缓存当中，然后会在 ConsumerNetworkClient\#poll 方法调用时才会被真正地执行发送。

fetcher.sendFetches() 在构建 FetchRequest 前，会对当前可拉取分区进行筛选，而这个也是决定多分区拉取消息规律的核心，后面我会讲到。

从 KafkaConsumer\#poll 方法源码可以看出来，其实 Kafka 消费者在拉取消息过程中，有两条线程在工作，其中用户主线程调用 pollForFetches 方法从缓存中获取消息消费，在获取消息后，会再调用 ConsumerNetworkClient\#poll 方法从 Broker 发送拉取请求，然后将拉取到的消息缓存到本地，这里为什么在拉取完消息后，会主动调用 ConsumerNetworkClient\#poll 方法呢？我想这里的目的是为了下次 poll 的时候可以立即从缓存中拉取消息。

pollForFetches 方法会调用 Fetcher\#fetchedRecords 方法从缓存中获取并解析消息：

*  public Map<TopicPartition, List<ConsumerRe

[5edeff5b69374ebaa86623fd56e68d1f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/329b26b97d22491ab837312ab89ec3b3.png
[8f21ef090da24e849d4ba0348a56b5bc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/50bc0647bbf444d38b8ae2039ec46b42.png