• Kafka 0.10.1.0 版本开始，KafkaConsumer 就变为了双线程的设计，即用户主线程和心跳线程
    • 用户主线程，就是你启动 Consumer 的那个线程
    • 心跳线程（Heartbeat Thread）只负责定期给对应的 Broker 机器发送心跳请求，以标识消费者应用的存活性（liveness）
  • Kafka Consumer 不是线程安全的,
    • 除了 KafkaConsumer#wakeup() ，其功能是来唤醒 Consumer

方案

  1. 消费者程序启动多个线程，每个线程维护专属的 KafkaConsumer 实例

    1. 每个线程负责完整的消息获取、消息处理流程

  2. 消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑

    1. 消息获取与消息处理过程解耦


image.png

代码实现

  1. public class KafkaConsumerRunner implements Runnable {
  2.      private final AtomicBoolean closed = new AtomicBoolean(false);
  3.      private final KafkaConsumer consumer;
  6.      public void run() {
  7.          try {
  8.              consumer.subscribe(Arrays.asList("topic"));
  9.              while (!closed.get()) {
  10.             ConsumerRecords records = 
  11.                 consumer.poll(Duration.ofMillis(10000));
  12.                  //  执行消息处理逻辑
  13.              }
  14.          } catch (WakeupException e) {
  15.              // Ignore exception if closing
  16.              if (!closed.get()) throw e;
  17.          } finally {
  18.              consumer.close();
  19.          }
  20.      }
  23.      // Shutdown hook which can be called from a separate thread
  24.      public void shutdown() {
  25.          closed.set(true);
  26.          consumer.wakeup();
  27.      }
  1. private final KafkaConsumer<String, String> consumer;
  2. private ExecutorService executors;
  3. ...
  6. private int workerNum = ...;
  7. executors = new ThreadPoolExecutor(
  8.     workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
  9.     new ArrayBlockingQueue<>(1000), 
  10.     new ThreadPoolExecutor.CallerRunsPolicy());
  13. ...
  14. while (true)  {
  15.     ConsumerRecords<String, String> records = 
  16.         consumer.poll(Duration.ofSeconds(1));
  17.     for (final ConsumerRecord record : records) {
  18.         executors.submit(new Worker(record));
  19.     }
  20. }
  21. ..