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

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

智能推荐

oracle 12c 集群安装后的检查_12c查看crs状态-程序员宅基地

文章浏览阅读1.6k次。安装配置gi、安装数据库软件、dbca建库见下：http://blog.csdn.net/kadwf123/article/details/784299611、检查集群节点及状态：[root@rac2 ~]# olsnodes -srac1 Activerac2 Activerac3 Activerac4 Active[root@rac2 ~]_12c查看crs状态

解决jupyter notebook无法找到虚拟环境的问题_jupyter没有pytorch环境-程序员宅基地

文章浏览阅读1.3w次，点赞45次，收藏99次。我个人用的是anaconda3的一个python集成环境，自带jupyter notebook，但在我打开jupyter notebook界面后，却找不到对应的虚拟环境，原来是jupyter notebook只是通用于下载anaconda时自带的环境，其他环境要想使用必须手动下载一些库：1.首先进入到自己创建的虚拟环境(pytorch是虚拟环境的名字)activate pytorch2.在该环境下下载这个库conda install ipykernelconda install nb__jupyter没有pytorch环境

国内安装scoop的保姆教程_scoop-cn-程序员宅基地

文章浏览阅读5.2k次，点赞19次，收藏28次。选择scoop纯属意外，也是无奈，因为电脑用户被锁了管理员权限，所有exe安装程序都无法安装，只可以用绿色软件，最后被我发现scoop，省去了到处下载XXX绿色版的烦恼，当然scoop里需要管理员权限的软件也跟我无缘了（譬如everything）。推荐添加dorado这个bucket镜像，里面很多中文软件，但是部分国外的软件下载地址在github，可能无法下载。以上两个是官方bucket的国内镜像，所有软件建议优先从这里下载。上面可以看到很多bucket以及软件数。如果官网登陆不了可以试一下以下方式。_scoop-cn

Element ui colorpicker在Vue中的使用_vue el-color-picker-程序员宅基地

文章浏览阅读4.5k次，点赞2次，收藏3次。首先要有一个color-picker组件 <el-color-picker v-model="headcolor"></el-color-picker>在data里面data() { return {headcolor: ’ #278add ’ //这里可以选择一个默认的颜色} }然后在你想要改变颜色的地方用v-bind绑定就好了，例如：这里的:sty..._vue el-color-picker

迅为iTOP-4412精英版之烧写内核移植后的镜像_exynos 4412 刷机-程序员宅基地

文章浏览阅读640次。基于芯片日益增长的问题，所以内核开发者们引入了新的方法，就是在内核中只保留函数，而数据则不包含，由用户（应用程序员）自己把数据按照规定的格式编写，并放在约定的地方，为了不占用过多的内存，还要求数据以根精简的方式编写。boot启动时，传参给内核，告诉内核设备树文件和kernel的位置，内核启动时根据地址去找到设备树文件，再利用专用的编译器去反编译dtb文件，将dtb还原成数据结构，以供驱动的函数去调用。firmware是三星的一个固件的设备信息，因为找不到固件，所以内核启动不成功。_exynos 4412 刷机

Linux系统配置jdk_linux配置jdk-程序员宅基地

文章浏览阅读2w次，点赞24次，收藏42次。Linux系统配置jdkLinux学习教程，Linux入门教程（超详细）_linux配置jdk

随便推点

matlab(4)：特殊符号的输入_matlab微米怎么输入-程序员宅基地

文章浏览阅读3.3k次，点赞5次，收藏19次。xlabel('\delta');ylabel('AUC');具体符号的对照表参照下图：_matlab微米怎么输入

C语言程序设计-文件(打开与关闭、顺序、二进制读写)-程序员宅基地

文章浏览阅读119次。顺序读写指的是按照文件中数据的顺序进行读取或写入。对于文本文件，可以使用fgets、fputs、fscanf、fprintf等函数进行顺序读写。在C语言中，对文件的操作通常涉及文件的打开、读写以及关闭。文件的打开使用fopen函数，而关闭则使用fclose函数。在C语言中，可以使用fread和fwrite函数进行二进制读写。‍ Biaoge 于2024-03-09 23:51发布阅读量：7 ️文章类型：【 C语言程序设计】在C语言中，用于打开文件的函数是____，用于关闭文件的函数是____。

Touchdesigner自学笔记之三_touchdesigner怎么让一个模型跟着鼠标移动-程序员宅基地

文章浏览阅读3.4k次，点赞2次，收藏13次。跟随鼠标移动的粒子以grid（SOP）为partical（SOP）的资源模板，调整后连接【Geo组合+point spirit（MAT)】，在连接【feedback组合】适当调整。影响粒子动态的节点【metaball(SOP)+force(SOP)】添加mouse in（CHOP)鼠标位置到metaball的坐标，实现鼠标影响。..._touchdesigner怎么让一个模型跟着鼠标移动

【附源码】基于java的校园停车场管理系统的设计与实现61m0e9计算机毕设SSM_基于java技术的停车场管理系统实现与设计-程序员宅基地

文章浏览阅读178次。项目运行环境配置：Jdk1.8 + Tomcat7.0 + Mysql + HBuilderX（Webstorm也行）+ Eclispe（IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持）。项目技术：Springboot + mybatis + Maven +mysql5.7或8.0+html+css+js等等组成，B/S模式 + Maven管理等等。环境需要1.运行环境：最好是java jdk 1.8，我们在这个平台上运行的。其他版本理论上也可以。_基于java技术的停车场管理系统实现与设计

Android系统播放器MediaPlayer源码分析_android多媒体播放源码分析时序图-程序员宅基地

文章浏览阅读3.5k次。前言对于MediaPlayer播放器的源码分析内容相对来说比较多，会从Java-&amp;gt;Jni-&amp;gt;C/C++慢慢分析，后面会慢慢更新。另外，博客只作为自己学习记录的一种方式，对于其他的不过多的评论。MediaPlayerDemopublic class MainActivity extends AppCompatActivity implements SurfaceHolder.Cal..._android多媒体播放源码分析时序图

java 数据结构与算法 ——快速排序法-程序员宅基地

文章浏览阅读2.4k次，点赞41次，收藏13次。java 数据结构与算法 ——快速排序法_快速排序法