Kafka Consumer API 的使用

2024-04-06 18:58
文章标签 使用 api kafka consumer

本文主要是介绍Kafka Consumer API 的使用,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

导读

Kafka具有两套消费者API:高级API、低级API。本文章将介绍两种API的区别以及使用时需要注意的地方。

低级API

1. 使用方法

  1. find leader broker
  2. build request
  3. fetch data
  4. identify leader change

2. 为什么要find leader

kafka在0.8版本后,引入replication机制。每个partition是有备份的,在某个broker出故障后,用户仍可以从其他备份中读取数据。消费者并不是并行从多个broker上获取同一个partition的数据,而是选举出一个leader broker,这个broker上的该partition将用于读写。其他的partition则复制leader broker上partition的数据,保持同步。

备份个数由创建topic时,replica-factory决定,当该参数为1时,表示不备份,大于1时,每个partition将有多个备份partition,且分布在不同的broker上。

3. 适合场合

  1. 消费起点需要设置
  2. 反复消费某一段数据

备注:笔者项目中有一个需求是,解析数据时,既可以支持从断点消费,又可以支持从当前位移消费。这个需求就使用了这个API实现功能。

高级API

1. 原理

高级API使用,围绕数据流工作,利用的低级API消费数据。用户不用关心leader broker、offset等问题。

2. 线程

使用高级API时,需要关注线程问题。

用户在使用高级API时,需要指定每个topic获取数据的线程数量。一个线程对应一个数据流。但是寻找主分区、创建流、设置offset这些过程中,高级API仍只有一个线程。只有当从partition中获取数据时,每个流才会产生一个fetchRunable的线程。

每个topic的线程数,最好设置为等于或者小于topic的partition个数。

3. Zookeeper

Kafka的使用需要Zookeeper有以下原因:

  1. 动态集群扩展。
  2. broker的注册,保存topic、partition元数据。
  3. consumer的注册。
  4. watcher的注册。

而高级API使用中,上述Zookeeper的作用全都用到了。首先,均衡(balance)partition和consumer时,需要两者信息。第二,kafka通过Watcher知道broker、topic、partition是否有变化。第三,kafka通过与zk通讯监控partition leader存活性.

另外,在笔者实验中发现,用kill指令杀死进程时,该进程中kafka消费者在zk中的注册信息可能并没有及时删除,如果马上拉起这个进程,将会可能出现消费者大于partition个数的情况。这种情况并不是必现,原因可能和kafka与zk的通讯时间有关系(会多查阅一点资料,验证猜想是否正确)

4. Rebalance

因为每个流,其实都需要指定数据来源的partition.每次创建线程,从partion中获取数据时,需要将同一个topic所有的partition和该group中消费该topic的所有线程合理分配,保证每一个partition只被一个线程消费。这个过程叫做balance,由高级API自动完成。

当发生以下三种情况的时候,会触发Kafka高级API的rebalance动作:

  1. 同一个group下,有新的消费者加入。
  2. 同一个group下,有topic的partition个数有变化。
  3. kafka API与zk的连接中断。

前两种情况比较好理解,重点讲第三种情况。我的理解是,zk超时或者断开后,kafka没有注册partition的信息,需要重新连接zk获取最新的注册信息,并根据新获取的信息进行线程、分区之间的分配和均衡。(个人理解,我会多查阅一些资料证实。)

zk超时后rebalance其实是很有可能不成功,并导致更多次的rebalance。原因是,如果kafka rebalance尝试的总时间(即尝试次数*每次尝试时间)小于zk超时时间,那么在zk连接失败重连之前,kafka的rebalance已经失败。这个原因可能会导致高级API不断的rebalance。而高级API默认设置参数,rebalance的尝试总时间是小于zk超时时间的,所以大家使用高级API时要根据实际情况处理这一点。



作者:君子月满楼
链接:https://www.jianshu.com/p/4d03ee74ad66
来源:简书
简书著作权归作者所有,任何形式的转载都请联系作者获得授权并注明出处。

这篇关于Kafka Consumer API 的使用的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/880579

相关文章

Spring Boot整合Redis注解实现增删改查功能(Redis注解使用)

《SpringBoot整合Redis注解实现增删改查功能(Redis注解使用)》文章介绍了如何使用SpringBoot整合Redis注解实现增删改查功能,包括配置、实体类、Repository、Se... 目录配置Redis连接定义实体类创建Repository接口增删改查操作示例插入数据查询数据删除数据更

使用python生成固定格式序号的方法详解

《使用python生成固定格式序号的方法详解》这篇文章主要为大家详细介绍了如何使用python生成固定格式序号,文中的示例代码讲解详细,具有一定的借鉴价值,有需要的小伙伴可以参考一下... 目录生成结果验证完整生成代码扩展说明1. 保存到文本文件2. 转换为jsON格式3. 处理特殊序号格式(如带圈数字)4

Java使用Swing生成一个最大公约数计算器

《Java使用Swing生成一个最大公约数计算器》这篇文章主要为大家详细介绍了Java使用Swing生成一个最大公约数计算器的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以了解一下... 目录第一步:利用欧几里得算法计算最大公约数欧几里得算法的证明情形 1:b=0情形 2:b>0完成相关代码第二步:加

Java中流式并行操作parallelStream的原理和使用方法

《Java中流式并行操作parallelStream的原理和使用方法》本文详细介绍了Java中的并行流(parallelStream)的原理、正确使用方法以及在实际业务中的应用案例,并指出在使用并行流... 目录Java中流式并行操作parallelStream0. 问题的产生1. 什么是parallelS

Linux join命令的使用及说明

《Linuxjoin命令的使用及说明》`join`命令用于在Linux中按字段将两个文件进行连接,类似于SQL的JOIN,它需要两个文件按用于匹配的字段排序,并且第一个文件的换行符必须是LF,`jo... 目录一. 基本语法二. 数据准备三. 指定文件的连接key四.-a输出指定文件的所有行五.-o指定输出

Linux jq命令的使用解读

《Linuxjq命令的使用解读》jq是一个强大的命令行工具,用于处理JSON数据,它可以用来查看、过滤、修改、格式化JSON数据,通过使用各种选项和过滤器,可以实现复杂的JSON处理任务... 目录一. 简介二. 选项2.1.2.2-c2.3-r2.4-R三. 字段提取3.1 普通字段3.2 数组字段四.

Linux kill正在执行的后台任务 kill进程组使用详解

《Linuxkill正在执行的后台任务kill进程组使用详解》文章介绍了两个脚本的功能和区别,以及执行这些脚本时遇到的进程管理问题,通过查看进程树、使用`kill`命令和`lsof`命令,分析了子... 目录零. 用到的命令一. 待执行的脚本二. 执行含子进程的脚本,并kill2.1 进程查看2.2 遇到的

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储:配置灵活性:编码示例引入依赖:配置ehcache.XML文件:配置

Java 虚拟线程的创建与使用深度解析

《Java虚拟线程的创建与使用深度解析》虚拟线程是Java19中以预览特性形式引入,Java21起正式发布的轻量级线程,本文给大家介绍Java虚拟线程的创建与使用,感兴趣的朋友一起看看吧... 目录一、虚拟线程简介1.1 什么是虚拟线程?1.2 为什么需要虚拟线程?二、虚拟线程与平台线程对比代码对比示例:三

k8s按需创建PV和使用PVC详解

《k8s按需创建PV和使用PVC详解》Kubernetes中,PV和PVC用于管理持久存储,StorageClass实现动态PV分配,PVC声明存储需求并绑定PV,通过kubectl验证状态,注意回收... 目录1.按需创建 PV(使用 StorageClass)创建 StorageClass2.创建 PV