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

精选留言(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

  

  