kafka多线程消费

Kafka consumer

多线程消费

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

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

• 每个线程维护一个KafkaConsumer，每个KafkaConsumer消费一个topic分区
• 单个KafkaConsumer实例统一拉取数据，交给多个worker线程进行处理

多Consumer

• 程序代码

importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.common.serialization.StringDeserializer;importjava.time.Duration;importjava.util.Arrays;importjava.util.Properties;publicclassMultiThreadConsumer{publicstaticvoidmain(String[] args){String brokers ="localhost:9092";String topic   ="topic_t40";String groupId ="app_q";int consumers  =2;for(int i =0;i < consumers;i++){finalConsumerRunnable consumer =newConsumerRunnable(brokers, groupId, topic,"thread"+ i);newThread(consumer).start();}}staticclassConsumerRunnableimplementsRunnable{privatefinalKafkaConsumer<String,String> consumer;privatevolatileboolean isRunning =true;privateString threadName ;publicConsumerRunnable(String brokers,String groupId,String topic,String threadName){Properties props =newProperties();
            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 =newKafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList(topic));this.threadName = threadName;}@Overridepublicvoidrun(){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线程

• 程序代码publicclassWorkerConsumer{privatestaticExecutorService executor =Executors.newFixedThreadPool(100);publicstaticvoidmain(String[] args){String topicName ="topic_t40";Properties props =newProperties(); 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 =newKafkaConsumer<>(props); consumer.subscribe(Arrays.asList(topicName));try{while(true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofSeconds(1));if(!records.isEmpty()){ executor.execute(newMessageHandler(records));}}}finally{ consumer.close();}}staticclassMessageHandlerimplementsRunnable{privateConsumerRecords<String,String> records;publicMessageHandler(ConsumerRecords<String,String> records){this.records = records;}@Overridepublicvoidrun(){ 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  10485760122721:07 00000000000000000000.index
-rw-r--r--  1 andy  wheel         0122619:49 00000000000000000000.log
-rw-r--r--  1 andy  wheel  10485756122721:07 00000000000000000000.timeindex
-rw-r--r--  1 andy  wheel         8122721:07 leader-epoch-checkpoint
-rw-r--r--  1 andy  wheel        43122619: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_*"));