极客时间已完结课程限时免费阅读

20 | 多线程开发消费者实例

20 | 多线程开发消费者实例-极客时间

20 | 多线程开发消费者实例

2019-07-18 胡夕 来自北京
《Kafka核心技术与实战》
课程介绍

讲述：胡夕

时长11:34大小10.59M

你好，我是胡夕。今天我们来聊聊 Kafka Java Consumer 端多线程消费的实现方案。
目前，计算机的硬件条件已经大大改善，即使是在普通的笔记本电脑上，多核都已经是标配了，更不用说专业的服务器了。如果跑在强劲服务器机器上的应用程序依然是单线程架构，那实在是有点暴殄天物了。不过，Kafka Java Consumer 就是单线程的设计，你是不是感到很惊讶。所以，探究它的多线程消费方案，就显得非常必要了。

Kafka Java Consumer 设计原理

在开始探究之前，我先简单阐述下 Kafka Java Consumer 为什么采用单线程的设计。了解了这一点，对我们后面制定多线程方案大有裨益。
谈到 Java Consumer API，最重要的当属它的入口类 KafkaConsumer 了。我们说 KafkaConsumer 是单线程的设计，严格来说这是不准确的。因为，从 Kafka 0.10.1.0 版本开始，KafkaConsumer 就变为了双线程的设计，即用户主线程和心跳线程。
所谓用户主线程，就是你启动 Consumer 应用程序 main 方法的那个线程，而新引入的心跳线程（Heartbeat Thread）只负责定期给对应的 Broker 机器发送心跳请求，以标识消费者应用的存活性（liveness）。引入这个心跳线程还有一个目的，那就是期望它能将心跳频率与主线程调用 KafkaConsumer.poll 方法的频率分开，从而解耦真实的消息处理逻辑与消费者组成员存活性管理。
不过，虽然有心跳线程，但实际的消息获取逻辑依然是在用户主线程中完成的。因此，在消费消息的这个层面上，我们依然可以安全地认为 KafkaConsumer 是单线程的设计。
其实，在社区推出 Java Consumer API 之前，Kafka 中存在着一组统称为 Scala Consumer 的 API。这组 API，或者说这个 Consumer，也被称为老版本 Consumer，目前在新版的 Kafka 代码中已经被完全移除了。
我之所以重提旧事，是想告诉你，老版本 Consumer 是多线程的架构，每个 Consumer 实例在内部为所有订阅的主题分区创建对应的消息获取线程，也称 Fetcher 线程。老版本 Consumer 同时也是阻塞式的（blocking），Consumer 实例启动后，内部会创建很多阻塞式的消息获取迭代器。但在很多场景下，Consumer 端是有非阻塞需求的，比如在流处理应用中执行过滤（filter）、连接（join）、分组（group by）等操作时就不能是阻塞式的。基于这个原因，社区为新版本 Consumer 设计了单线程 + 轮询的机制。这种设计能够较好地实现非阻塞式的消息获取。
除此之外，单线程的设计能够简化 Consumer 端的设计。Consumer 获取到消息后，处理消息的逻辑是否采用多线程，完全由你决定。这样，你就拥有了把消息处理的多线程管理策略从 Consumer 端代码中剥离的权利。
另外，不论使用哪种编程语言，单线程的设计都比较容易实现。相反，并不是所有的编程语言都能够很好地支持多线程。从这一点上来说，单线程设计的 Consumer 更容易移植到其他语言上。毕竟，Kafka 社区想要打造上下游生态的话，肯定是希望出现越来越多的客户端的。

多线程方案

了解了单线程的设计原理之后，我们来具体分析一下 KafkaConsumer 这个类的使用方法，以及如何推演出对应的多线程方案。
首先，我们要明确的是，KafkaConsumer 类不是线程安全的 (thread-safe)。所有的网络 I/O 处理都是发生在用户主线程中，因此，你在使用过程中必须要确保线程安全。简单来说，就是你不能在多个线程中共享同一个 KafkaConsumer 实例，否则程序会抛出 ConcurrentModificationException 异常。
当然了，这也不是绝对的。KafkaConsumer 中有个方法是例外的，它就是 wakeup()，你可以在其他线程中安全地调用 KafkaConsumer.wakeup() 来唤醒 Consumer。
鉴于 KafkaConsumer 不是线程安全的事实，我们能够制定两套多线程方案。
1.消费者程序启动多个线程，每个线程维护专属的 KafkaConsumer 实例，负责完整的消息获取、消息处理流程。如下图所示：
2.消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑。获取消息的线程可以是一个，也可以是多个，每个线程维护专属的 KafkaConsumer 实例，处理消息则交由特定的线程池来做，从而实现消息获取与消息处理的真正解耦。具体架构如下图所示：
总体来说，这两种方案都会创建多个线程，这些线程都会参与到消息的消费过程中，但各自的思路是不一样的。
我们来打个比方。比如一个完整的消费者应用程序要做的事情是 1、2、3、4、5，那么方案 1 的思路是粗粒度化的工作划分，也就是说方案 1 会创建多个线程，每个线程完整地执行 1、2、3、4、5，以实现并行处理的目标，它不会进一步分割具体的子任务；而方案 2 则更细粒度化，它会将 1、2 分割出来，用单线程（也可以是多线程）来做，对于 3、4、5，则用另外的多个线程来做。
这两种方案孰优孰劣呢？应该说是各有千秋。我总结了一下这两种方案的优缺点，我们先来看看下面这张表格。
接下来，我来具体解释一下表格中的内容。
我们先看方案 1，它的优势有 3 点。
实现起来简单，因为它比较符合目前我们使用 Consumer API 的习惯。我们在写代码的时候，使用多个线程并在每个线程中创建专属的 KafkaConsumer 实例就可以了。
多个线程之间彼此没有任何交互，省去了很多保障线程安全方面的开销。
由于每个线程使用专属的 KafkaConsumer 实例来执行消息获取和消息处理逻辑，因此，Kafka 主题中的每个分区都能保证只被一个线程处理，这样就很容易实现分区内的消息消费顺序。这对在乎事件先后顺序的应用场景来说，是非常重要的优势。
说完了方案 1 的优势，我们来看看这个方案的不足之处。
每个线程都维护自己的 KafkaConsumer 实例，必然会占用更多的系统资源，比如内存、TCP 连接等。在资源紧张的系统环境中，方案 1 的这个劣势会表现得更加明显。
这个方案能使用的线程数受限于 Consumer 订阅主题的总分区数。我们知道，在一个消费者组中，每个订阅分区都只能被组内的一个消费者实例所消费。假设一个消费者组订阅了 100 个分区，那么方案 1 最多只能扩展到 100 个线程，多余的线程无法分配到任何分区，只会白白消耗系统资源。当然了，这种扩展性方面的局限可以被多机架构所缓解。除了在一台机器上启用 100 个线程消费数据，我们也可以选择在 100 台机器上分别创建 1 个线程，效果是一样的。因此，如果你的机器资源很丰富，这个劣势就不足为虑了。
每个线程完整地执行消息获取和消息处理逻辑。一旦消息处理逻辑很重，造成消息处理速度慢，就很容易出现不必要的 Rebalance，从而引发整个消费者组的消费停滞。这个劣势你一定要注意。我们之前讨论过如何避免 Rebalance，如果你不记得的话，可以回到专栏第 17 讲复习一下。
下面我们来说说方案 2。
与方案 1 的粗粒度不同，方案 2 将任务切分成了消息获取和消息处理两个部分，分别由不同的线程处理它们。比起方案 1，方案 2 的最大优势就在于它的高伸缩性，就是说我们可以独立地调节消息获取的线程数，以及消息处理的线程数，而不必考虑两者之间是否相互影响。如果你的消费获取速度慢，那么增加消费获取的线程数即可；如果是消息的处理速度慢，那么增加 Worker 线程池线程数即可。
不过，这种架构也有它的缺陷。
它的实现难度要比方案 1 大得多，毕竟它有两组线程，你需要分别管理它们。
因为该方案将消息获取和消息处理分开了，也就是说获取某条消息的线程不是处理该消息的线程，因此无法保证分区内的消费顺序。举个例子，比如在某个分区中，消息 1 在消息 2 之前被保存，那么 Consumer 获取消息的顺序必然是消息 1 在前，消息 2 在后，但是，后面的 Worker 线程却有可能先处理消息 2，再处理消息 1，这就破坏了消息在分区中的顺序。还是那句话，如果你在意 Kafka 中消息的先后顺序，方案 2 的这个劣势是致命的。
方案 2 引入了多组线程，使得整个消息消费链路被拉长，最终导致正确位移提交会变得异常困难，结果就是可能会出现消息的重复消费。如果你在意这一点，那么我不推荐你使用方案 2。

实现代码示例

讲了这么多纯理论的东西，接下来，我们来看看实际的实现代码大概是什么样子。毕竟，就像 Linus 说的：“Talk is cheap, show me the code!”
我先跟你分享一段方案 1 的主体代码：
public class KafkaConsumerRunner implements Runnable {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final KafkaConsumer consumer;
public void run() {
try {
consumer.subscribe(Arrays.asList("topic"));
while (!closed.get()) {
ConsumerRecords records =
consumer.poll(Duration.ofMillis(10000));
// 执行消息处理逻辑
}
} catch (WakeupException e) {
// Ignore exception if closing
if (!closed.get()) throw e;
} finally {
consumer.close();
}
}
// Shutdown hook which can be called from a separate thread
public void shutdown() {
closed.set(true);
consumer.wakeup();
}
这段代码创建了一个 Runnable 类，表示执行消费获取和消费处理的逻辑。每个 KafkaConsumerRunner 类都会创建一个专属的 KafkaConsumer 实例。在实际应用中，你可以创建多个 KafkaConsumerRunner 实例，并依次执行启动它们，以实现方案 1 的多线程架构。
对于方案 2 来说，核心的代码是这样的：
private final KafkaConsumer<String, String> consumer;
private ExecutorService executors;
...
private int workerNum = ...;
executors = new ThreadPoolExecutor(
workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(1000),
new ThreadPoolExecutor.CallerRunsPolicy());
...
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
for (final ConsumerRecord record : records) {
executors.submit(new Worker(record));
}
}
..
这段代码最重要的地方是最后一行：当 Consumer 的 poll 方法返回消息后，由专门的线程池来负责处理具体的消息。调用 poll 方法的主线程不负责消息处理逻辑，这样就实现了方案 2 的多线程架构。

小结

总结一下，今天我跟你分享了 Kafka Java Consumer 多线程消费的实现方案。我给出了比较通用的两种方案，并介绍了它们各自的优缺点以及代码示例。我希望你能根据这些内容，结合你的实际业务场景，实现适合你自己的多线程架构，真正做到举一反三、融会贯通，彻底掌握多线程消费的精髓，从而在日后实现更宏大的系统。

开放讨论

今天我们讨论的都是多线程的方案，可能有人会说，何必这么麻烦，我直接启动多个 Consumer 进程不就得了？那么，请你比较一下多线程方案和多进程方案，想一想它们各自的优劣之处。
欢迎写下你的思考和答案，我们一起讨论。如果你觉得有所收获，也欢迎把文章分享给你的朋友。
分享给需要的人，Ta购买本课程，你将得20元
生成海报并分享

赞 10

提建议

© 版权归极客邦科技所有，未经许可不得传播售卖。 页面已增加防盗追踪，如有侵权极客邦将依法追究其法律责任。
上一篇
19 | CommitFailedException异常怎么处理？
下一篇
21 | Java 消费者是如何管理TCP连接的?
unpreview
 写留言

精选留言(69)

  • yhh
    2019-07-18
    希望老师能讲讲方案2下线程池怎么管理和提交位移！！
    共 19 条评论
    77
  • 小生向北
    2019-07-18
    能够用多线程解决的就不要用多进程，毕竟资源有限。方案2的讲解还是太浅了，同希望老师能针对方案2详细讲解一下！方案2里面在异步线程里提交offset，每个线程自己提交自己的，如果中间有offset提交失败，后面的offset又提交成功了咋办呢？而且每个线程都自己提交consumer.commit()就意味着要在多个线程里面使用consumer，如文中所说，这种情况是要报CME错误的，那究竟该如何正确的提交呢，有没有最佳实践呀？
    共 4 条评论
    40
  • james
    2019-07-18
    方案2最核心的如何commit老师没有说，难道只能启用自动提交吗？我觉得可以用Cyclicbarrier来实现线程池执行完毕后，由consumer来commit，不用countdownlatch因为它只能记录一次，而cb可以反复用，或者用forkjoin方式，总之要等待多线程都处理完才能commit，风险就是某个消息处理太慢回导致整体都不能commit，而触发rebalance以及重复消费，而重复消费我用布隆过滤器来解决
    共 4 条评论
    24
  • QQ怪
    2019-07-18
    老师能否加餐spring-kafka相关知识
    共 2 条评论
    25
  • 注定非凡
    2019-11-05
    A ：Kafka Java Consumer是单线程设计原理。 （1）在Kafka从0.10.1.0版本开始，KafkaConsumer就变成双线程设计即：用户主线程和心跳线程。 （2）主线程是指：启动Consumer应用程序main方法的那个线程，而新引入的心跳线程只负责定期给对应的Broker机器发送心跳请求，以标识消费者应用的存活性。 （2）老版本中有Scala Consumer的API，是多线程架构的，每个Consumer实例在内部为所有订阅的主题分区创建对应消息获取线程，也称为Fetcher线程。老版本Consumer同时也是阻塞式的（blocking），Consumer实例启动后，内部会创建很多阻塞式的消息迭代器。 （3）在很多场景下，Consumer端是有非阻塞需求的，如流处理应用中执行过滤（filter），连接（join），分组（group by）等操作时就不能是阻塞式的。 所以，新版本Consumer设计了单线程+轮询的机制。这种设计能够较好的实现非阻塞式的消息获取。 B ：单线程设计优点 （1）单线程可以较好的实现如在流处理应用中执行过滤（filter），连接（join）,分组（group by）等操作。 （2）单线程能够简化Consumer端设计。Consumer端获取到消息后，处理消息的逻辑是否采用多线程，由自己决定。 （3）单线程设计在很多种编程中都比较易于实现，编译社区移植。 C ：多线程方案 （1）KafkaConsumer类不是线程安全的（thread-safe）。所有的网络I/O处理都是发生在用户主线程中，所以不能在多线程中共享同一个KafkaConsumer实例，否则程序会抛ConcurrentModificationException异常。 （2）方案一： 消费者程序启动多个线程，每个线程维护专属的KafkaConsumer实例，负责完整的消息获取，消息处理流程。 优点： 方便实现，速度快，无线程间交互开销，易于维护分区的消息顺序 缺点： 占用更多的系统资源，线程数受限于主题分区数，扩展性差。线程自己处理消息容易超时，进而引发Rebalance。 （3）方案二： 消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑。获取消息的线程可以是多个，每个线程维护专属的KafkaConsumer实例，处理消息则交由特定的线程池来做。 优点： 可独立扩展消费获取线程数和worker线程数，伸缩性好 缺点： 难以维护分区内的消息消费顺序，处理链路拉长，不易于位移提交管理，实现难度高。
    展开
    21
  • calljson
    2019-07-18
    希望老师能对比spring-kafka源码，关于多线程管理consumer谢谢
    22
  • 千屿
    2019-07-20
    最近用spring cloud做了一个kafka可靠消息微服务组件，有兴趣的朋友可以看看 ，消费端是多线程模型，消费线程和业务执行分离，使用了mongodb(分片+副本集) 存储消息发送的链路，对发送失败的消息做了补偿机制。https://gitee.com/huacke/mq-kafka，有问题可以联系我。
    20
  • 寂静欢喜
    2019-11-27
    老师 想问下 心跳线程是和主线程分开的，那么 第一种方案中，主线程阻塞，又怎么会导致超时Rebalance呢？

    作者回复: 应该这么说，心跳线程会定期地检查前端线程是否卡住了，一旦发现卡住了，便会发起主动离组。

    共 3 条评论
    20
  • 玉剑冰锋
    2019-07-18
    Kafka重启时间比较长，每次重启一台差不多四五十分钟，日志保存12个小时，每台数据量差不多几个T，想请教一下老师有什么可以优化的参数吗？

    作者回复: 有可能是要加载的日志段数据太多导致的，可以增加num.recovery.threads.per.data.dir的值

    共 4 条评论
    18
  • 来碗绿豆汤
    2019-07-22
    对于第二种方案，可以添加一个共享的队列，消费线程消费完一个记录，就写入队列，然后主线程可以读取这个队列，然后依次提交小号的offset
    共 1 条评论
    14
  • yic
    2020-02-12
    老师，关于方案2中的做法，位移提交是有重复消费消息和丢失数据的风险的，有没有什么好的实践呀？

    作者回复: 最好的办法就是自己完全实现一套多线程+管理offset的方案，就像Spark Streaming和Flink做的那样。有兴趣的话可以阅读以下Flink中Kafka Connector的源代码：）

    共 2 条评论
    8
  • z.l
    2019-07-20
    请教个问题，如果使用方案1，一个consumer group订阅了2个topic，每个topic都是24个分区，此时最大线程数可以设置为24还是48？

    作者回复: 理论上是48，但实际上这么多线程反而是开销，可以采用多进程的方式

    8
  • KEEPUP
    2019-07-19
    希望老师讲一下sparkstreaming 消费kafka 消息的情况
    7
  • 飞翔
    2019-12-04
    老师 想问一个方案1 谁负责分配线程给每个partition呀 我看您的code 只是没产生一个线程去消费一个主题 如果我有4个parition 那么我产生4个线程来消费这个主题，他会自动均匀分配嘛

    作者回复: “谁负责分配线程给每个partition呀” --- leader consumer负责分配。 会均匀分配的，这是kafka consumer代码保证的

    7
  • 张洋
    2020-05-19
    老师如果当前consumer group下的consumer instance 只分配了当前主题的一个分区是不是意味着 当前也只能是一个线程来消费消息了

    作者回复: 取决于consumer instance是线程还是进程。通常情况下如果consumer instance是进程的话，还是可以使用多个线程来消费这个获取到的数据。

    5
  • 高志强
    2019-12-25
    老师我现在用Php多进程消费，一个topic 130个分区，我是不是该启动130个进程去消费，目前启动64个进程，但消费能力上不去，消息积压量有几十万了，怎么才能提高消费能力呢

    作者回复: 可以考虑单个进程下再开多线程的方式来增强消费能力，不必一味考虑多进程的方案

    共 4 条评论
    4
  • Hale
    2019-12-24
    如果只有一个broker,一个consumer 一个分区，上面的consumer 组成一个组，一个topic 当consumer 卡住时，协调器会将消费者踢出消费组，进行重新分区分配，但只有一个消费者，那消费者就不能接受到数据了，怎样实现消费者重连

    作者回复: 消费者会自动重连的，如果重连失败，说明网络有问题

    4
  • YWH
    2019-12-16
    老师，想请教消费者的一个问题... 我们的业务场景是这样的：建立一个服务接收 http 请求、根据传入的参数（topic）从 Kafka 指定 topic 拉取一定数量的消息后返回。但 Kafka 的消费者是要保持轮询的，不然就只能每次建立消费者、获取分区/加入群组、请求数据后关闭消费者（但这样效率很低）。 请问有什么比较好又可靠的实现方法吗？谢谢~

    作者回复: 持续消费一部分消息缓存到本地，http接口从本地读取消息。如果长时间不拉取，consumer先pause消费

    4
  • 胡家鹏
    2019-10-23
    老师及各位朋友好，问下两个问题1.上面的代码怎么没有消费位移提交，难道是设置的自动提交位移吗？2.consumer.wakeup什么时候使用，来解决什么问题呢？

    作者回复: 1. 您指哪段代码？另外如果设置了enable.auto.commit=true或没有显式设置enable.auto.commit=false，就是自动提交 2. wakeup主要用于唤醒polling中的consumer实例。如果你使用了多线程（即把KafkaConsumer实例用于单独的线程），你需要有能力在另一个线程中“中断”KafkaConsumer所在实例的执行。wakeup就是用这个的

    共 4 条评论
    4
  • Xiao
    2019-07-18
    胡老师，第二种方案我觉得还有个问题就是如果是自动提交，那就会容易出现消息丢失，因为异步消费消息，如果worker线程有异常，主线程捕获不到异常，就会造成消息丢失，这个是不是还需要做补偿机制；如果是手动提交，那offer set也会有可能丢失，消息重复消费，消息重复还好处理，做幂等就行。
    4