热门标签
热门文章
- 153个Python库，你必须要试试_python常用库打包下载
- 2（附源码）SSM人事管理系统的设计和实现 计算机毕设48260_人事管理系统的后台数据库设计
- 3【Calcite源码学习】SqlNode方言转换_calcite sqlnode
- 4阿里通义千问 Qwen1.5-MoE 开源！推理训练最佳实践教程来啦_qwen moe 开源
- 5如何在python代码中使用HTTP代理IP_python ip代理
- 6详细教程 - 从零开发 Vue 鸿蒙harmonyOS应用 第一节_鸿蒙deveco vue
- 7《系统集成项目管理》第十二章 项目沟通管理和干系人管理_沟通管理和干系人管理的区别和联系
- 8ESP32 http 请求_vscode-idf 获取时间
- 9一文带你学会使用YOLO及Opencv完成图像及视频流目标检测（上）|附源码_yolo模型对视频识别的原理和对图片识别的原理
- 10Springboot源码学习-启动流程_bootstrapregistryinitializer
当前位置: article > 正文
kafka多线程消费
作者：小蓝xlanll | 2024-04-14 02:57:30
赞
踩
kafka多线程消费
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(); } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 测试结果
多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()); }); } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
-
测试结果
方法对比
多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
- 1
- 2
- 3
在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
- 1
- 2
- 3
- 4
- 5
- 6
- 7
当多个consumer 或 consumer group需要同时提交位移信息时，kafka会根据每个消费者的group.id 做hash取模运算，从而将位移数据负载到不同的分区上。
订阅主题
kafka 消费者处理支持常规的topic列表进行订阅之外，还支持基于正则表达式订阅topic，代码实现分别如下：
- Topic列表订阅
consumer.subscribe(Arrays.asList("hello","world"));
- 1
- 基于正则表达是订阅
consumer.subscribe(Pattern.compile("topic_*"));
- 1
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/419828
推荐阅读
相关标签
