Kafka【十三】消费者消费消息的偏移量

2024-09-07 06:44

本文主要是介绍Kafka【十三】消费者消费消息的偏移量,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

偏移量offset是消费者消费数据的一个非常重要的属性。默认情况下,消费者如果不指定消费主题数据的偏移量,那么消费者启动消费时,无论当前主题之前存储了多少历史数据,消费者只能从连接成功后当前主题最新的数据偏移位置读取,而无法读取之前的任何数据。如果想要获取之前的数据,就需要设定配置参数或指定数据偏移量。

【1】起始偏移量

在消费者的配置中,我们可以增加偏移量相关参数auto.offset.reset,用于从最开始获取主题数据。

在这里插入图片描述
参数取值有3个:latest、earliest、none。

① earliest

earliest:对于同一个消费者组,从头开始消费。就是说如果这个topic有历史消息存在,现在新启动了一个消费者组,且auto.offset.reset=earliest,那将会从头开始消费(未提交偏移量的场合)。
在这里插入图片描述

② latest

latest:对于同一个消费者组,消费者只能消费到连接topic后,新产生的数据(未提交偏移量的场合)。

在这里插入图片描述

none:生产环境不使用。

【2】 指定偏移量消费

除了从最开始的偏移量或最后的偏移量读取数据以外,Kafka还支持从指定的偏移量的位置开始消费数据。
在这里插入图片描述

【3】偏移量提交

生产环境中,消费者可能因为某些原因或故障重新启动消费,那么如果不知道之前消费数据的位置,重启后再消费,就可能重复消费(earliest)或漏消费(latest)。所以Kafka提供了保存消费者偏移量的功能,而这个功能需要由消费者进行提交操作。这样消费者重启后就可以根据之前提交的偏移量进行消费了。

注意,一旦消费者提交了偏移量,那么kafka会优先使用提交的偏移量进行消费。此时,auto.offset.reset参数是不起作用的。

① 自动提交

所谓的自动提交就是消费者消费完数据后,无需告知kafka当前消费数据的偏移量,而是由消费者客户端API周期性地将消费的偏移量提交到Kafka中。这个周期默认为5000ms,可以通过配置进行修改。

在这里插入图片描述

② 手动提交

基于时间周期的偏移量提交是我们无法控制的,一旦参数设置的不合理或单位时间内数据量消费的很多,却没有来及的自动提交,那么数据就会重复消费。所以Kafka也支持消费偏移量的手动提交,也就是说当消费者消费完数据后,自行通过API进行提交。

不过为了考虑效率和安全,kafka同时提供了异步提交和同步提交两种方式供我们选择。注意:需要禁用自动提交auto.offset.reset=false,才能开启手动提交

异步提交

向Kafka发送偏移量offset提交请求后,就可以直接消费下一批数据,因为无需等待kafka的提交确认,所以无法知道当前的偏移量一定提交成功,所以安全性比较低,但相对消费性能会提高。

在这里插入图片描述

同步提交

必须等待Kafka完成offset提交请求的响应后,才可以消费下一批数据。一旦提交失败,会进行重试处理,尽可能保证偏移量提交成功(但是依然可能因为意外情况导致提交请求失败)。此种方式消费效率比较低,但是安全性高。
在这里插入图片描述

【4】偏移量保存

由于消费者在消费消息的时候可能会由于各种原因而断开消费,当重新启动消费者时我们需要让它接着上次消费的位置offset继续消费,因此消费者需要实时的记录自己以及消费的位置。

0.90版本之前,这个信息是记录在zookeeper内的,在0.90之后的版本,offset保存在__consumer_offsets这个topic内。

每个consumer会定期将自己消费分区的offset提交给kafka内部topic:__consumer_offsets,提交过去的时候,key是consumerGroupId+topic+分区号

在这里插入图片描述

value就是当前offset的值,kafka会定期清理topic里的消息,最后就保留最新的那条数据。

在这里插入图片描述
因为__consumer_offsets可能会接收高并发的请求,kafka默认给其分配50个分区(可以通过offsets.topic.num.partitions设置),均匀分配到Kafka集群的多个Broker中。Kafka采用hash(consumerGroupId) % __consumer_offsets主题的分区数来计算我们的偏移量提交到哪一个分区。

因为偏移量也是保存到主题中的,所以保存的过程和生产者生产数据的过程基本相同。

【5】消费者事务

无论偏移量使用自动提交还是,手动提交,特殊场景中数据都有可能会出现重复消费。

在这里插入图片描述
如果提前提交偏移量,再处理业务,又可能出现数据丢失的情况。在这里插入图片描述

对于单独的Consumer来讲,事务保证会比较弱,尤其是无法保证提交的信息被精确消费。主要原因就是消费者可以通过偏移量访问信息,而不同的数据文件生命周期不同,同一事务的信息可能会因为重启导致被删除的情况。

所以一般情况下,想要完成kafka消费者端的事务处理,需要将数据消费过程和偏移量提交过程进行原子性绑定。也就是说数据处理完了,必须要保证偏移量正确提交,才可以做下一步的操作。如果偏移量提交失败,那么数据就恢复成处理之前的效果。

对于生产者事务而言,消费者消费的数据也会受到限制。默认情况下,消费者只能消费到生产者提交的数据,也就是未提交完成的数据,消费者是看不到的。

这篇关于Kafka【十三】消费者消费消息的偏移量的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3

ActiveMQ—消息特性(延迟和定时消息投递)

ActiveMQ消息特性:延迟和定时消息投递(Delay and Schedule Message Delivery) 转自:http://blog.csdn.net/kimmking/article/details/8443872 有时候我们不希望消息马上被broker投递出去,而是想要消息60秒以后发给消费者,或者我们想让消息没隔一定时间投递一次,一共投递指定的次数。。。 类似

java线程深度解析(五)——并发模型(生产者-消费者)

http://blog.csdn.net/Daybreak1209/article/details/51378055 三、生产者-消费者模式     在经典的多线程模式中,生产者-消费者为多线程间协作提供了良好的解决方案。基本原理是两类线程,即若干个生产者和若干个消费者,生产者负责提交用户请求任务(到内存缓冲区),消费者线程负责处理任务(从内存缓冲区中取任务进行处理),两类线程之

JavaSE(十三)——函数式编程(Lambda表达式、方法引用、Stream流)

函数式编程 函数式编程 是 Java 8 引入的一个重要特性,它允许开发者以函数作为一等公民(first-class citizens)的方式编程,即函数可以作为参数传递给其他函数,也可以作为返回值。 这极大地提高了代码的可读性、可维护性和复用性。函数式编程的核心概念包括高阶函数、Lambda 表达式、函数式接口、流(Streams)和 Optional 类等。 函数式编程的核心是Lambda

Java消息队列:RabbitMQ与Kafka的集成与应用

Java消息队列:RabbitMQ与Kafka的集成与应用 大家好,我是微赚淘客返利系统3.0的小编,是个冬天不穿秋裤,天冷也要风度的程序猿! 在现代的分布式系统中,消息队列是实现系统间通信、解耦和提高可扩展性的重要组件。RabbitMQ和Kafka是两个广泛使用的消息队列系统,它们各有特点和优势。本文将介绍如何在Java应用中集成RabbitMQ和Kafka,并展示它们的应用场景。 消息队

Kafka (快速)安装部署

文章目录 1、软件下载&配置环境1_JDK安装2_Zookeeper安装3_Kafka安装 2、单机安装1_配置主机名和IP映射2_单机Kafka配置 3、集群安装1_配置主机名和IP的映射关系2_时钟同步3_Zookeeper配置信息4_集群Kafka配置 4、kafka的其他脚本命令 1、软件下载&配置环境 下面的操作无论是单机部署还是分布式集群环境下都是通用的。 准

Cmake之3.0版本重要特性及用法实例(十三)

简介: CSDN博客专家、《Android系统多媒体进阶实战》一书作者 新书发布:《Android系统多媒体进阶实战》🚀 优质专栏: Audio工程师进阶系列【原创干货持续更新中……】🚀 优质专栏: 多媒体系统工程师系列【原创干货持续更新中……】🚀 优质视频课程:AAOS车载系统+AOSP14系统攻城狮入门视频实战课 🚀 人生格言: 人生从来没有捷径,只有行动才是治疗恐惧

Kafka 分布式消息系统详细介绍

Kafka 分布式消息系统 一、Kafka 概述1.1 Kafka 定义1.2 Kafka 设计目标1.3 Kafka 特点 二、Kafka 架构设计2.1 基本架构2.2 Topic 和 Partition2.3 消费者和消费者组2.4 Replica 副本 三、Kafka 分布式集群搭建3.1 下载解压3.1.1 上传解压 3.2 修改 Kafka 配置文件3.2.1 修改zookeep

Android 友盟消息推送集成遇到的问题

友盟消息推送遇到的问题 集成友盟消息推送,步骤根据提供的技术文档接入便可。可是当你集成到项目中去的时候,可能并不是一帆风顺就搞定,因为你项目里面是可能集成了其他的sdk(比如支付宝,微信,七鱼等等三方的sdk)。那么这个时候,再加上友盟的消息推送sdk集成可能就会出现问题。 问题清单 友盟消息推送sdk和支付宝sdk冲突问题 后台配置了消息推送,也显示发送成功,但是手机没有收到消息通知

十三、我们应当怎样做需求分析:查询报表分析

在我以往的用例分析中,使用这样格式的用例模式,对于大多数业务操作流程来说是得心应手的,但对于有些功能来说总感觉不对劲。感觉不对劲的,就是那些查询、汇总与报表功能。对于这部分功能,需要我们描述的不是什么操作流程,而更重要的是那些数据项、数据来源、报表格式、数据链接,以及使用者、使用频率的说明。而这些,在以往的用例说明格式中统统都没有,怎么办呢?俗话说“东西是死的人是活的”,把我们的用例格式改改吧。