Kafka consumer

多线程消费

kafka 消费者对象 - KafkaConsumer是非线程安全的。与KafkaProducer不同，KafkaProducer是线程安全的，因为开发者可以在多个线程中放心地使用同一个KafkaProducer实例。

但是对于消费者而言，由于它是非线程安全的，因此用户无法直接在多个线程中直接共享同一个KafkaConsumer实例。对应kafka 多线程消费给出两种解决方案：

• 每个线程维护一个KafkaConsumer，每个KafkaConsumer消费一个topic分区

  

• 单个KafkaConsumer实例统一拉取数据，交给多个worker线程进行处理

  

多Consumer

• 程序代码

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class MultiThreadConsumer {

    public static void main(String[] args) {
        String brokers = "localhost:9092";
        String topic   = "topic_t40";
        String groupId = "app_q";
        int consumers  = 2;

        for(int i = 0;i < consumers;i++){
            final ConsumerRunnable consumer = new ConsumerRunnable(brokers, groupId, topic, "thread" + i);
            new Thread(consumer).start();
        }
    }

    static class ConsumerRunnable implements Runnable{

        private final KafkaConsumer<String,String> consumer;

        private volatile boolean isRunning = true;

        private String threadName ;

        public ConsumerRunnable(String brokers,String groupId,String topic,String threadName) {
            Properties props = new Properties();
            props.put("bootstrap.servers", brokers);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,   StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList(topic));

            this.threadName = threadName;
        }

        @Override
        public void run() {
            try {
                while (isRunning) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                    records.forEach(record -> {
                        System.out.println(this.threadName + " : Message received " + record.value() + ", partition " + record.partition());
                    });
                }
            }finally {
                consumer.close();
            }
        }
    }
}

• 测试结果

多Work线程

• 程序代码

  public class WorkerConsumer {
  
      private static ExecutorService executor = Executors.newFixedThreadPool(100);
  
      public static void main(String[] args){
  
          String topicName = "topic_t40";
  
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "app_w");
          props.put("client.id", "client_02");
          props.put("enable.auto.commit", true);
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,   StringDeserializer.class.getName());
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  
          KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
          consumer.subscribe(Arrays.asList(topicName));
          try {
              while (true) {
                  ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                  if(!records.isEmpty()){
                      executor.execute(new MessageHandler(records));
                  }
              }
          }finally {
              consumer.close();
          }
      }
  
      static class MessageHandler implements Runnable{
  
          private ConsumerRecords<String, String> records;
  
          public MessageHandler(ConsumerRecords<String, String> records) {
              this.records = records;
          }
  
          @Override
          public void run() {
              records.forEach(record -> {
                  System.out.println(" 开始处理消息: " + record.value() + ", partition " + record.partition());
              });
          }
      }
  }
  

• 测试结果

方法对比

多Consumer

• 优势 - 实现简单；速度较快 无线程切换，方便位移管理，易于维护分区间消息消费顺序
• 缺点 - socket连接大；consumer的数量受限于topic的分区数，扩展性差；

多Work线程

• 优势 - 消息获取与消息处理解耦；可独立扩展消费者数量和工作线程数量，伸缩性好
• 缺点 - 难以维护分区消息处理的有序性，位移管理困难

__consumer_offsets

之前提到过，消费者通过拉取模式从broker中拉取数据，每次消费成功后，消费者记录自身消费位移，当服务重启后，默认从最后的位移位置开始拉取最新的数据。那么消费者是如何记录自身的位移的呢？

__consumer_offset是kafka自行创建的一个内部topic，因此开发者不可以删除该topic，它的目的是存储Kafka 消费者的偏移量。consumer_offset是管理所有消费者的偏移量的一个主题。

# 查看kafka配置文件日志路径
more config/server.properties | grep log.dirs
log.dirs=/tmp/node0/kafka-logs

在kafka的日志目录中，可以看到由**__consumer_offsets**开头的带数字序号的50个文件夹，表示该topic有50个分区。进入任意文件夹，发现他跟正常的topic文件差不多，里面至少包含了2个index索引文件，一个日志文件

ls -ll /tmp/node0/kafka-logs/__consumer_offsets-1
total 16
-rw-r--r--  1 andy  wheel  10485760 12 27 21:07 00000000000000000000.index
-rw-r--r--  1 andy  wheel         0 12 26 19:49 00000000000000000000.log
-rw-r--r--  1 andy  wheel  10485756 12 27 21:07 00000000000000000000.timeindex
-rw-r--r--  1 andy  wheel         8 12 27 21:07 leader-epoch-checkpoint
-rw-r--r--  1 andy  wheel        43 12 26 19:49 partition.metadata00000000000000000000.log	00000000000000000000.timeindex	leader-epoch-checkpoint		partition.metadata

当多个consumer 或 consumer group需要同时提交位移信息时，kafka会根据每个消费者的group.id 做hash取模运算，从而将位移数据负载到不同的分区上。

订阅主题

kafka 消费者处理支持常规的topic列表进行订阅之外，还支持基于正则表达式订阅topic，代码实现分别如下：

• Topic列表订阅

consumer.subscribe(Arrays.asList("hello","world"));

• 基于正则表达是订阅

consumer.subscribe(Pattern.compile("topic_*"));