Kafka从上手到实践 - 实践真知：Kafka Java Consumer | 凌云时刻-程序员宅基地

凌云时刻 · 技术

导读：这一节来看看如何使用Java编写Kafka Consumer。

作者 | 计缘

来源 | 凌云时刻（微信号：linuxpk）

Java Consumer

首先创建Consumer需要的配置信息，最基本的有五个信息：

Kafka集群的地址。
发送的Message中Key的序列化方式。
发送的Message中Value的序列化方式。
指定Consumer Group。

指定拉取Message范围的策略。

Properties properties = new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "IP:Port");
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer_group_1");
properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // earliest, none

然后传入上面实例化好的配置信息，实例化Consumer：

KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

然后通过Consumer的subscribe(Collection<String> topics)方法订阅Topic：

consumer.subscribe(Arrays.asList("first_topic"));

最后获取Topic里的Message，将Message信息输出到日志中：

while(true) {
	ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
	for(ConsumerRecord<String, String> record : records) {
		logger.info("Key: " + record.key() + ", Value: " + record.value());
		logger.info("Partition: " + record.partition() + ", Offset: " + record.offset());
	}
}

Consumer的poll(Duration timeout)方法可以设置获取数据的时间间隔，同时回忆一下在之前Consumer章节的Consumer Poll Options小节中，说过关于Consumer获取Message的四个配置项，都可以在Properties里进行设置。

启动Java Consumer后，在控制台可以看到如下信息：

[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 2.0.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 3402a8361b734732
[main] INFO org.apache.kafka.clients.Metadata - Cluster ID: 4nh_0r5iQ_KsR_Fzf1HTGg
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Discovered group coordinator IP:9092 (id: 2147483647 rack: null)
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Revoking previously assigned partitions []
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] (Re-)joining group
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Successfully joined group with generation 1
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Setting newly assigned partitions [first_topic-0, first_topic-1, first_topic-2]
[main] INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-1, groupId=consumer_group_1] Resetting offset for partition first_topic-0 to offset 23.
[main] INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-1, groupId=consumer_group_1] Resetting offset for partition first_topic-1 to offset 24.
[main] INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-1, groupId=consumer_group_1] Resetting offset for partition first_topic-2 to offset 21.

在上面的信息中，可以看到Setting newly assigned partitions [first_topic-0, first_topic-1, first_topic-2]这句话，说明当前这个Consumer会获取first_topic这个Topic中全部Partition中的Message。

如果我们再启动一个Consumer，这个Consumer和第一个在同一个组里，看看会有什么输出信息：

[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 2.0.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 3402a8361b734732
[main] INFO org.apache.kafka.clients.Metadata - Cluster ID: 4nh_0r5iQ_KsR_Fzf1HTGg
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Discovered group coordinator IP:9092 (id: 2147483647 rack: null)
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Revoking previously assigned partitions []
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] (Re-)joining group
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Successfully joined group with generation 2
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Setting newly assigned partitions [first_topic-2]

可以看到新启动的Consumer会输出Setting newly assigned partitions [first_topic-2]这句话，说明新的这个Consumer只会获取first_topic这个Topic的一个Partition中的Message。

再回去看看第一个Consumer的控制台：

[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Attempt to heartbeat failed since group is rebalancing
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Revoking previously assigned partitions [first_topic-0, first_topic-1, first_topic-2]
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] (Re-)joining group
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Successfully joined group with generation 2
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1, groupId=consumer_group_1] Setting newly assigned partitions [first_topic-0, first_topic-1]

第一个Consumer新输出在控制台中的信息很关键，首先看到Attempt to heartbeat failed since group is rebalancing这句话，说明Kafka会自动重新给Consumer Group里的Consumer分配Topic的Partition。

再看Setting newly assigned partitions [first_topic-0, first_topic-1]这句，说明第一个Consumer不会再获取first_topic-2这个Partition里的Message了。这也印证了在Consumer章节的Consumer Group小节里讲过的概念。

Java Consumer

Java Consumer with Assign and Seek

如果我们有一个临时的Consumer，不想加入任何一个Consumer Group，而且需要指定Topic的Partition，以及指定从哪个Message Offset开始获取数据，怎么办？所幸，Kafka提供了这样的API。

首先我们在实例化配置信息时，就不需要指定Consumer Group了：

Properties properties = new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstant.BOOTSTRAP_SERVER);
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // earliest, none

然后实例化TopicPartition，指定Topic和Partition序号。使用Consumer的assign(Collection<TopicPartition> partitions)方法，分配给该Consumer：

TopicPartition topicPartition = new TopicPartition("first_topic", 0);
consumer.assign(Arrays.asList(topicPartition));

再然后指定Message Offset：

long offset = 21L;
consumer.seek(topicPartition, offset);

运行该Consumer，可以看到如下输出信息：

[main] INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-1, groupId=] Fetch offset 21 is out of range for partition first_topic-0, resetting offset
[main] INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-1, groupId=] Resetting offset for partition first_topic-0 to offset 22.
[main] INFO com.devtalking.jacefu.kafka.tutorial.ConsumerDemoAssignSeek - Key: null, Value: hello world!
[main] INFO com.devtalking.jacefu.kafka.tutorial.ConsumerDemoAssignSeek - Partition: 0, Offset: 22

如果我们使用Consumer Group CLI查看，会发现这种操作其实也是临时创建了一个Consumer Group：

root@iZ2ze2booskait1cxxyrljZ:~# kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list

consumer_group_1
KMOffsetCache-iZ2ze2booskait1cxxyrljZ

小结

这一章节带大家实践如何使用Kafka提供的API编写Java Consumer。上一节和这一节主要介绍了Kafka Java Client（Producer和Consumer）的使用方式，相比Kafka CLI，Java Client在实际的开发中可能使用的更加频繁，希望能给使用Java语言的小伙伴们带来帮助。

END

往期精彩文章回顾

Kafka Java Producer

Kafka CLI：Reseting Offset & Config CLI

Kafka CLI：Consumer CLI & Producer CLI

Kafka CLI：Topic CLI & Producer CLI

Kafka从上手到实践 - 实践真知：搭建单机Kafka

Kafka从上手到实践 - 庖丁解牛：Consumer

Kafka从上手到实践 - 庖丁解牛：Producer

Kafka从上手到实践 - 庖丁解牛：Partition

Kafka从上手到实践 - 庖丁解牛：Topic & Broker

Kafka从上手到实践 - 初步认知：MQ系统

长按扫描二维码关注凌云时刻

每日收获前沿技术与科技洞见

本文链接：https://blog.csdn.net/bjchenxu/article/details/107036676

原作者删帖不实内容删帖广告或垃圾文章投诉

智能推荐

2022黑龙江最新建筑八大员（材料员）模拟考试试题及答案_料账的试题-程序员宅基地

文章浏览阅读529次。百分百题库提供建筑八大员（材料员）考试试题、建筑八大员（材料员）考试预测题、建筑八大员（材料员）考试真题、建筑八大员（材料员）证考试题库等,提供在线做题刷题，在线模拟考试，助你考试轻松过关。310项目经理部应编制机械设备使用计划并报（）审批。A监理单位B企业C建设单位D租赁单位答案：B311对技术开发、新技术和新工艺应用等情况进行的分析和评价属于（）。A人力资源管理考核B材料管理考核C机械设备管理考核D技术管理考核答案：D312建筑垃圾和渣土._料账的试题

chatgpt赋能python：Python自动打开浏览器的技巧-程序员宅基地

文章浏览阅读614次。本文由chatgpt生成，文章没有在chatgpt生成的基础上进行任何的修改。以上只是chatgpt能力的冰山一角。作为通用的Aigc大模型，只是展现它原本的实力。对于颠覆工作方式的ChatGPT，应该选择拥抱而不是抗拒，未来属于“会用”AI的人。AI职场汇报智能办公文案写作效率提升教程专注于AI+职场+办公方向。下图是课程的整体大纲下图是AI职场汇报智能办公文案写作效率提升教程中用到的ai工具。_python自动打开浏览器

Linux中安装JDK-RPM_linux 安装jdk rpm-程序员宅基地

文章浏览阅读545次。Linux中安装JDK-RPM方式_linux 安装jdk rpm

net高校志愿者管理系统-73371，计算机毕业设计（上万套实战教程，赠送源码）-程序员宅基地

文章浏览阅读25次。免费领取项目源码，请关注赞收藏并私信博主，谢谢-高校志愿者管理系统主要功能模块包括页、个人资料（个人信息。修改密码）、公共管理（轮播图、系统公告）、用户管理（管理员、志愿用户）、信息管理（志愿资讯、资讯分类）、活动分类、志愿活动、报名信息、活动心得、留言反馈，采取面对对象的开发模式进行软件的开发和硬体的架设，能很好的满足实际使用的需求，完善了对应的软体架设以及程序编码的工作，采取SQL Server 作为后台数据的主要存储单元，采用Asp.Net技术进行业务系统的编码及其开发，实现了本系统的全部功能。

小米宣布用鸿蒙了吗,小米OV对于是否采用鸿蒙保持沉默，原因是中国制造需要它们...-程序员宅基地

文章浏览阅读122次。原标题：小米OV对于是否采用鸿蒙保持沉默，原因是中国制造需要它们目前华为已开始对鸿蒙系统大规模宣传，不过中国手机四强中的另外三家小米、OPPO、vivo对于是否采用鸿蒙系统保持沉默，甚至OPPO还因此而闹出了一些风波，对此柏铭科技认为这是因为中国制造当下需要小米OV几家继续将手机出口至海外市场。 2020年中国制造支持中国经济渡过了艰难的一年，这一年中国进出口贸易额保持稳步增长的势头，成为全球唯一..._小米宣布用鸿蒙系统

Kafka Eagle_kafka eagle git-程序员宅基地

文章浏览阅读1.3k次。1.Kafka Eagle实现kafka消息监控的代码细节是什么?2.Kafka owner的组成规则是什么？3.怎样使用SQL进行kafka数据预览？4.Kafka Eagle是否支持多集群监控？1.概述在《Kafka 消息监控 - Kafka Eagle》一文中，简单的介绍了 Kafka Eagle这款监控工具的作用，截图预览，以及使用详情。今天_kafka eagle git

随便推点

Eva.js是什么（互动小游戏开发）-程序员宅基地

文章浏览阅读1.1k次，点赞29次，收藏19次。Eva.js 是一个专注于开发互动游戏项目的前端游戏引擎。：Eva.js 提供开箱即用的游戏组件供开发人员立即使用。是的，它简单而优雅！：Eva.js 由高效的运行时和渲染管道 (Pixi.JS) 提供支持，这使得释放设备的全部潜力成为可能。：得益于 ECS（实体-组件-系统）架构，你可以通过高度可定制的 API 扩展您的需求。唯一的限制是你的想象力！_eva.js

OC学习笔记-Objective-C概述和特点_objective-c特点及应用领域-程序员宅基地

文章浏览阅读1k次。Objective-C概述Objective-C是一种面向对象的计算机语言，1980年代初布莱德.考斯特在其公司Stepstone发明Objective-C，该语言是基于SmallTalk-80。1988年NeXT公司发布了OC，他的开发环境和类库叫NEXTSTEP， 1994年NExt与Sun公司发布了标准的NEXTSTEP系统，取名openStep。1996_objective-c特点及应用领域

STM32学习笔记6：TIM基本介绍_stm32 tim寄存器详解-程序员宅基地

文章浏览阅读955次，点赞20次，收藏16次。TIM（Timer）定时器定时器可以对输入的时钟进行计数，并在计数值达到设定值时触发中断16位计数器、预分频器、自动重装寄存器的时基单元，在 72MHz 计数时钟下可以实现最大 59.65s 的定时，59.65s65536×65536×172MHz59.65s65536×65536×721MHz不仅具备基本的定时中断功能，而且还包含内外时钟源选择、输入捕获、输出比较、编码器接口、主从触发模式等多种功能。_stm32 tim寄存器详解

前端基础语言HTML、CSS 和 JavaScript 学习指南_艾编程学习资料-程序员宅基地

文章浏览阅读1.5k次。对于任何有兴趣学习前端 Web 开发的人来说，了解 HTML、CSS 和JavaScript 之间的区别至关重要。这三种前端语言都是您访问过的每个网站的用户界面构建块。而且，虽然每种语言都有不同的功能重点，但它们都可以共同创建令人兴奋的交互式网站，让用户保持参与。因此，您会发现学习所有三种语言都很重要。如果您有兴趣从事前端开发工作，可以通过多种方式学习这些语言——在艾编程就可以参与到学习当中来。在本文中，我们将回顾每种语言的特征、它们如何协同工作以及您可以在哪里学习它们。HTML vs C._艾编程学习资料

三维重构（10）：PCL点云配准_局部点云与全局点云配准-程序员宅基地

文章浏览阅读2.8k次。点云配准主要针对点云的：不完整、旋转错位、平移错位。因此要得到完整点云就需要对局部点云进行配准。为了得到被测物体的完整数据模型，需要确定一个合适的坐标系变换，将从各个视角得到的点集合并到一个统一的坐标系下形成一个完整的数据点云，然后就可以方便地进行可视化，这就是点云数据的配准。点云配准技术通过计算机技术和统计学规律，通过计算机计算两个点云之间的错位，也就是把在不同的坐标系下的得到的点云进行坐标变..._局部点云与全局点云配准

python零基础学习书-Python零基础到进阶必读的书藉：Python学习手册pdf免费下载-程序员宅基地

文章浏览阅读273次。提取码：0oorGoogle和YouTube由于Python的高可适应性、易于维护以及适合于快速开发而采用它。如果你想要编写高质量、高效的并且易于与其他语言和工具集成的代码，《Python学习手册：第4 版》将帮助你使用Python快速实现这一点，不管你是编程新手还是Python初学者。本书是易于掌握和自学的教程，根据作者Python专家Mark Lutz的著名培训课程编写而成。《Python学习..._零基础学pythonpdf电子书