Kafka核心技术与实战
胡夕
人人贷计算平台部总监,Apache Kafka Contributor
立即订阅
8408 人已学习
课程目录
已完结 46 讲
0/4登录后,你可以任选4讲全文学习。
开篇词 (1讲)
开篇词 | 为什么要学习Kafka?
免费
Kafka入门 (5讲)
01 | 消息引擎系统ABC
02 | 一篇文章带你快速搞定Kafka术语
03 | Kafka只是消息引擎系统吗?
04 | 我应该选择哪种Kafka?
05 | 聊聊Kafka的版本号
Kafka的基本使用 (3讲)
06 | Kafka线上集群部署方案怎么做?
07 | 最最最重要的集群参数配置(上)
08 | 最最最重要的集群参数配置(下)
客户端实践及原理剖析 (14讲)
09 | 生产者消息分区机制原理剖析
10 | 生产者压缩算法面面观
11 | 无消息丢失配置怎么实现?
12 | 客户端都有哪些不常见但是很高级的功能?
13 | Java生产者是如何管理TCP连接的?
14 | 幂等生产者和事务生产者是一回事吗?
15 | 消费者组到底是什么?
16 | 揭开神秘的“位移主题”面纱
17 | 消费者组重平衡能避免吗?
18 | Kafka中位移提交那些事儿
19 | CommitFailedException异常怎么处理?
20 | 多线程开发消费者实例
21 | Java 消费者是如何管理TCP连接的?
22 | 消费者组消费进度监控都怎么实现?
深入Kafka内核 (5讲)
23 | Kafka副本机制详解
24 | 请求是怎么被处理的?
25 | 消费者组重平衡全流程解析
26 | 你一定不能错过的Kafka控制器
27 | 关于高水位和Leader Epoch的讨论
管理与监控 (12讲)
28 | 主题管理知多少?
29 | Kafka动态配置了解下?
30 | 怎么重设消费者组位移?
31 | 常见工具脚本大汇总
32 | KafkaAdminClient:Kafka的运维利器
33 | Kafka认证机制用哪家?
34 | 云环境下的授权该怎么做?
35 | 跨集群备份解决方案MirrorMaker
36 | 你应该怎么监控Kafka?
37 | 主流的Kafka监控框架
38 | 调优Kafka,你做到了吗?
39 | 从0搭建基于Kafka的企业级实时日志流处理平台
高级Kafka应用之流处理 (3讲)
40 | Kafka Streams与其他流处理平台的差异在哪里?
41 | Kafka Streams DSL开发实例
42 | Kafka Streams在金融领域的应用
结束语 (1讲)
结束语 | 以梦为马,莫负韶华!
特别放送 (2讲)
加餐 | 搭建开发环境、阅读源码方法、经典学习资料大揭秘
用户故事 | 黄云:行百里者半九十
Kafka核心技术与实战
登录|注册

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

胡夕 2019-07-18
你好,我是胡夕。今天我们来聊聊 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 方法的频率分开,从而解耦真实的消息处理逻辑与消费者组成员存活性管理。
取消
完成
0/1000字
划线
笔记
复制
© 版权归极客邦科技所有,未经许可不得传播售卖。 页面已增加防盗追踪,如有侵权极客邦将依法追究其法律责任。
该试读文章来自付费专栏《Kafka核心技术与实战》,如需阅读全部文章,
请订阅文章所属专栏。
立即订阅
登录 后留言

精选留言(38)

  • yhh
    希望老师能讲讲方案2下线程池怎么管理和提交位移!!
    2019-07-18
    7
    38
  • QQ怪
    老师能否加餐spring-kafka相关知识
    2019-07-18
    16
  • calljson
    希望老师能对比spring-kafka源码,关于多线程管理consumer谢谢
    2019-07-18
    12
  • 小生向北
    能够用多线程解决的就不要用多进程,毕竟资源有限。方案2的讲解还是太浅了,同希望老师能针对方案2详细讲解一下!方案2里面在异步线程里提交offset,每个线程自己提交自己的,如果中间有offset提交失败,后面的offset又提交成功了咋办呢?而且每个线程都自己提交consumer.commit()就意味着要在多个线程里面使用consumer,如文中所说,这种情况是要报CME错误的,那究竟该如何正确的提交呢,有没有最佳实践呀?
    2019-07-18
    8
  • james
    方案2最核心的如何commit老师没有说,难道只能启用自动提交吗?我觉得可以用Cyclicbarrier来实现线程池执行完毕后,由consumer来commit,不用countdownlatch因为它只能记录一次,而cb可以反复用,或者用forkjoin方式,总之要等待多线程都处理完才能commit,风险就是某个消息处理太慢回导致整体都不能commit,而触发rebalance以及重复消费,而重复消费我用布隆过滤器来解决
    2019-07-18
    6
  • 千屿
    最近用spring cloud做了一个kafka可靠消息微服务组件,有兴趣的朋友可以看看 ,消费端是多线程模型,消费线程和业务执行分离,使用了mongodb(分片+副本集) 存储消息发送的链路,对发送失败的消息做了补偿机制。https://gitee.com/huacke/mq-kafka,有问题可以联系我。
    2019-07-20
    4
  • 注定非凡
    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线程数,伸缩性好
    缺点:
    难以维护分区内的消息消费顺序,处理链路拉长,不易于位移提交管理,实现难度高。
    2019-11-05
    2
  • 来碗绿豆汤
    对于第二种方案,可以添加一个共享的队列,消费线程消费完一个记录,就写入队列,然后主线程可以读取这个队列,然后依次提交小号的offset
    2019-07-22
    2
  • KEEPUP
    希望老师讲一下sparkstreaming 消费kafka 消息的情况
    2019-07-19
    2
  • Xiao
    胡老师,第二种方案我觉得还有个问题就是如果是自动提交,那就会容易出现消息丢失,因为异步消费消息,如果worker线程有异常,主线程捕获不到异常,就会造成消息丢失,这个是不是还需要做补偿机制;如果是手动提交,那offer set也会有可能丢失,消息重复消费,消息重复还好处理,做幂等就行。
    2019-07-18
    2
  • 玉剑冰锋
    Kafka重启时间比较长,每次重启一台差不多四五十分钟,日志保存12个小时,每台数据量差不多几个T,想请教一下老师有什么可以优化的参数吗?

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

    2019-07-18
    3
    2
  • 开水
    方案一用在需要精确控制消费数量的方案里,比如访问量这种日志什么的。
    方案二可以把后面处理通过数据库key做成幂等操作,用在实时处理需要随时增减消费能力的业务上面。
    2019-07-19
    1
  • 丘壑
    对于老师说的第二种多线程处理的方案,我本人觉得在消息量很大的系统中比较常用,只是在使用的时候很担心出现异常后的数据问题,数据应该怎么找回,这块对消费异常设计难度较大,请老师可以考虑分享下这块的手动提交位移及异常处理的经验
    2019-07-19
    1
  • 飞翔
    老师 想问一个方案1 谁负责分配线程给每个partition呀 我看您的code 只是没产生一个线程去消费一个主题 如果我有4个parition 那么我产生4个线程来消费这个主题,他会自动均匀分配嘛

    作者回复: “谁负责分配线程给每个partition呀” --- leader consumer负责分配。

    会均匀分配的,这是kafka consumer代码保证的

    2019-12-04
  • 寂静欢喜
    老师 想问下 心跳线程是和主线程分开的,那么 第一种方案中,主线程阻塞,又怎么会导致超时Rebalance呢?

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

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

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

    2019-10-23
  • miwucc
    方案2可以实现自己的分区线程池执行方法,不过对于位移管理还是存在风险,不能保证一定处理成功
    2019-09-16
  • Geek_b809ff
    胡老师,请教一个问题。用命令行消费是ok的,但是用API消费,在调用了consumer.poll(1000) 方法后就没任何反应了,请问有可能是什么问题?具体实现代码如下,用了线程池
    public void start() {
            try {
                int threadCoreNumber = 5;
                int threadMaxNumber = 10;
                //启用线程池
                executor = new ThreadPoolExecutor(threadCoreNumber, threadMaxNumber, 1L, TimeUnit.MINUTES,
                        new ArrayBlockingQueue<Runnable>(500), new ThreadPoolExecutor.CallerRunsPolicy());
                Thread thread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            while (true) {
                                //从kafka中读取消息
                                ConsumerRecords<String, String> records = consumer.poll(1000);
                                //自动提交
                                for (ConsumerRecord<String, String> record : records) {
                                    logger.info(String.format("[consumer][thread:%s] receive message from [Topic:%s -> partition:%s -> offset:%s], message key:%s ,value:%s",
                                            Thread.currentThread().getName(), record.topic(), record.partition(), record.offset(), record.key(), record.value()));
                                    executor.submit(new SaleMngConsumer(record));
                                }
                            }
                        } catch (Exception e) {
                            logger.info("djfs",e);
                            //ignore if shutdown
                        }finally {
                            logger.info("kafka consumer is close ......");
                            consumer.close();
                        }
                    }
                });
                thread.start();
            } catch (Exception e) {
                executor.shutdown();
            }
        }

    作者回复: 同一个consumer实例是在多个线程间共享的吗?KafkaConsumer不是线程安全的,你应该不被允许这么做的

    2019-09-11
  • Geek_b809ff
    老师,我是Kafka小白,一直有一个疑问,就是Kafka为什么能实现削峰限流的效果。比如说,我现在有一个业务场景的并发是每秒1000,这个时候数据库已经处理不过来了,如果这时候选择加入Kafka,由于Kafka可以让更多的消息发过来,如果消费者端也开启多线程去处理的话, 那数据库岂不是更加处理不过来了吗。请老师解答,谢谢!

    作者回复: 针对你的场景,可以先将数据缓存在Kafka中,然后对下游的consumer进行限流

    2019-09-01
  • z.l
    有个疑问请教下,使用批量消费+手动提交的方式,如果中间某一条消息由于代码bug抛出异常,导致没有提交消费位移,这种情况是不是这个分区的消费进度就一直停滞不前了?并且这个消费者实例一直在重复消费这一批消息?

    作者回复: 如果碰到异常你可以中断consumer消费,修复之后再重新上线

    2019-08-30
收起评论
38
返回
顶部