为什么kafka 需要 subscribe 的 group.id?我们是否需要使用 commitSync 手动提交偏移量?
目录
- 一、为什么需要带有 subscribe 的 group.id
- 二、我们需要使用commitSync手动提交偏移量吗?
- 三、如果我想手动提交偏移量,该怎么做?
一、为什么需要带有 subscribe 的 group.id
- 消费概念:
Kafka 使用消费者组的概念来实现主题的并行消费 - 每条消息都将在每个消费者组中传递一次,无论该组中实际有多少个消费者。所以 group 参数是强制性的,如果没有组,Kafka 将不知道如何对待订阅同一主题的其他消费者。 - 偏移量:
每当我们启动一个消费者时,它都会加入一个消费者组,然后根据该消费者组中的其他消费者数量,为其分配要读取的分区。对于这些分区,它会检查列表读取偏移量是否已知,如果找到,它将从这一点开始读取消息。如果没有找到偏移量,则参数 auto.offset.reset 控制是从分区中最早的消息还是从最新的消息开始读取。
二、我们需要使用commitSync手动提交偏移量吗?
-
是否需要手动提交偏移?
是否需要提交偏移量取决于作为参数 enable.auto.commit 选择的值。默认情况下,此设置为 true,这意味着消费者将定期自动提交其偏移量(由auto.commit.interval.ms 决定提交的频率)。如果将其设置为 false,那么将需要自己提交偏移量。这种默认行为可能也是导致很多发现 kafka 总是从最新的开始消费的原因,由于偏移量是自动提交的,因此它将使用该偏移量。 -
有没有办法从头开始重播消息?
如果想每次都从头开始读取,可以调用seekToBeginning,如果不带参数调用,它将重置为所有订阅分区中的第一条消息,或者仅重置您传入的那些分区。 -
seekToBeginning
查找每个给定分区的第一个偏移量。poll(long) 该函数延迟计算,仅在调用或时才查找所有分区中的第一个偏移量position(TopicPartition)。如果未提供分区,则查找所有当前分配的分区的第一个偏移量。public class MyListener implements ConsumerSeekAware {...@Overridepublic void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {callback.seekToBeginning(assignments.keySet());}} -
有没有办法从最后开始重播消息?
有的,可以使用 seekToEnd() 查找所有分配的分区到最后。或者使用 seekToTimestamp(long time)- 查找所有分配的分区到该时间戳表示的偏移量。public class MyListener extends AbstractConsumerSeekAware {@KafkaListener(...)void listn(...) {...} }public class SomeOtherBean {MyListener listener;...void someMethod() {this.listener.seekToTimestamp(System.currentTimeMillis - 60_000);}}
三、如果我想手动提交偏移量,该怎么做?
-
1、禁用自动提交
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); -
提交方法
对于手动提交,KafkaConsumers提供了两种方法,即 commitSync() 和 commitAsync()。commitSync()是一个阻塞调用,在偏移量成功提交后返回,commitAsync()则立即返回。如果想知道提交是否成功,可以为回调处理程序 ( OffsetCommitCallback) 提供一个方法参数。请注意,在两次提交调用中,消费者都会提交最新poll()调用的偏移量。
举个例子:假设一个分区主题有一个消费者并且最后一次调用poll()返回偏移量为 4、5、6 的消息。提交时,偏移量 6 将被提交,因为这是消费者客户端跟踪的最新偏移量。
同时,commitSync() 和 commitAsync() 都允许更多地控制我们想要提交的偏移量:如果你使用允许你指定的相应重载,那么Map<TopicPartition, OffsetAndMetadata>消费者将仅提交指定的偏移量(即,映射可以包含分配的分区的任何子集) ,并且指定的偏移量可以为任意值)。 -
同步提交:
阻塞线程,直到提交成功或遇到不可恢复的错误(在这种情况下,它被抛出给调用者)while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());consumer.commitSync();} }对于 for 循环中的每次迭代,只有在consumer.commitSync()成功返回或因抛出异常而中断后,代码才会移至下一次迭代。
-
异步提交:
是一种非阻塞方法。调用它不会阻塞线程。相反,它将继续处理以下指令,无论最终是成功还是失败。while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());consumer.commitAsync(callback);} }对于 for 循环中的每次迭代,无论consumer.commitAsync()最终会发生什么,代码都会移至下一次迭代。并且,提交的结果将由定义的回调函数处理。
-
权衡:延迟与数据一致性
1、如果必须确保数据一致性,请选择commitSync(),因为它将确保在执行任何进一步操作之前,你将知道偏移量提交是成功还是失败。但由于它是同步和阻塞的,你将花费更多的时间来等待提交完成,这会导致高延迟。
2、如果可以接受某些数据不一致并希望具有低延迟,请选择commitAsync(),因为它不会等待完成。相反,它只会发出提交请求并稍后处理来自 Kafka 的响应(成功或失败),同时代码将继续执行。
相关文章:
为什么kafka 需要 subscribe 的 group.id?我们是否需要使用 commitSync 手动提交偏移量?
目录 一、为什么需要带有 subscribe 的 group.id二、我们需要使用commitSync手动提交偏移量吗?三、如果我想手动提交偏移量,该怎么做? 一、为什么需要带有 subscribe 的 group.id 消费概念: Kafka 使用消费者组的概念来实现主题的…...
什么是Web应用程序防火墙,WAF与其他网络安全工具差异在哪?
一、什么是Web 应用程序防火墙 (WAF) ? WAF软件产品被广泛应用于保护Web应用程序和网站免受威胁或攻击,它通过监控用户、应用程序和其他互联网来源之间的流量,有效防御跨站点伪造、跨站点脚本(XSS攻击)、SQL注入、DDo…...
打家劫舍 II——力扣213
动规 int robrange(vector<int>& nums, int start, int end){int first=nums[start]...
动手学深度学习—卷积神经网络LeNet(代码详解)
1. LeNet LeNet由两个部分组成: 卷积编码器:由两个卷积层组成;全连接层密集块:由三个全连接层组成。 每个卷积块中的基本单元是一个卷积层、一个sigmoid激活函数和平均汇聚层;每个卷积层使用55卷积核和一个sigmoid激…...
腾讯面经总结
最近在准备面试,看了很多大厂的面经,抽空将腾讯面试的题目整理了一下,希望对大家有所帮助~ 一面 1、mysql索引结构? 2、redis持久化策略? 3、zookeeper节点类型说一下; 4、zookeeper选举机制ÿ…...
matlab机器人工具箱基础使用
资料:https://blog.csdn.net/huangjunsheng123/article/details/110630665 用vscode直接看工具箱api代码比较方便,代码说明很多 一、模型设置 1、基础效果 %采用机器人工具箱进行正逆运动学验证 a[0,-0.3,-0.3,0,0,0];%DH参数 d[0.05,0,0,0.06,0.05,…...
利用WonderLeak进行内存泄露检测【一】
1、下载地址: WonderLeak - Visual Studio Marketplace https://www.relyze.com/ 2、WonderLeak支持vs2017 2019扩展,或者单独启动 3、https://www.relyze.com/docs/wonderleak/help/w/overview/msvc_extension1.png 4、对于二进制程序来说支持以下…...
二刷LeetCode--155. 最小栈(C++版本),思维题
思路:本题需要使用两个栈,一个就是正常栈,执行出入操作,另一个栈只负责将对应的最小值进行保存即可.每次入栈的时候,最小值栈的栈顶也需要入栈元素,不过这个元素是最小值,那么就需要进行比较,因此在getmin()的时候只需要将最小值栈的栈顶元素弹出即可.初始化的时候只需要将最小…...
进程的状态与转换
进程在其生命周期内,由于系统中各进程之间的相互制约及系统的运行环境的变化,使得进程的状态也在不断地发生变化。通常进程有以下5种状态,前三种是基础讷航的基本状态 1)运行态。进程正在处理机上运行。在单处理机机中࿰…...
用MariaDB创建数据库,SQL练习,MarialDB安装和使用
前言:MariaDB数据库管理系统是MySQL的一个分支,主要由开源社区在维护,采用GPL授权许可 MariaDB的目的是完全兼容MySQL,包括API和命令行,使之能轻松成为MySQL的代替品。在存储引擎方面,使用XtraDB来代替MySQ…...
【Docker】 使用Docker-Compose 搭建基于 WordPress 的博客网站
引 本文将使用流行的博客搭建工具 WordPress 搭建一个私人博客站点。部署过程中使用到了 Docker 、MySQL 。站点搭建完成后经行了发布文章的体验。 WordPress WordPress 是一个广泛使用的开源内容管理系统(CMS),用于构建和管理网站、博客和…...
Hlang社区-前端社区宣传首页实现
文章目录 前言页面结构固定钉头部轮播JS特效完整代码总结前言 这里的话,博主其实也是今年参与考研的大军之一,所以的话,是抽空去完成这个项目的,当然这个项目的肯定是可以在较短的时间内完成的。 那么废话不多说,昨天也是干到1点多,把这个首页写出来了。先看看看效果吧:…...
【LeetCode-Medium】833. 字符串中的查找与替换
题目链接 833. 字符串中的查找与替换 标签 字符串 步骤 Step1. 初始化 ans[]: for (int i 0; i < s.length(); i) { // 初始化ansans[i] s[i]; }Step2. 根据 index, source, target 查找;如果找到,那么将 ans[i] 更改为 target&am…...
数据结构中公式前中后缀表达式-二叉树应用
目录 数据结构中公式前中后缀表达式-二叉树应用 数据结构中公式前中后缀表达式-二叉树应用 什么是前缀表达式、中缀表达式、后缀表达式 前缀表达式、中缀表达式、后缀表达式,是通过树来存储和计算表达式的三种不同方式 以如下公式为例 通过树来存储该公式&#x…...
Visual Studio 2022连接远程系统进行C/C++开发
Visual Studio被称为是宇宙最强IDE,以前开发Linux C/C服务器程序,基本上都是在Windows上使用VS编写跨平台的C/C代码,然后先在VS中编译、链接、调试,然后在Linux下编译、链接,再针对Linux下的特定代码进行调试。后面Vis…...
TiDB数据库从入门到精通系列之二:TiDB数据库的简介
TiDB数据库从入门到精通系列之二:TiDB数据库的简介 一、TiDB数据库的简介二、五大核心特性三、四大核心应用场景四、TiDB数据库与MySQL数据库的兼容性 一、TiDB数据库的简介 TiDB是开源分布式关系型数据库,是一款同时支持在线事务处理与在线分析处理 (H…...
opencv视频截取每一帧并保存为图片python代码CV2实现练习
当涉及到视频处理时,Python中的OpenCV库提供了强大的功能,可以方便地从视频中截取每一帧并将其保存为图片。这是一个很有趣的练习,可以让你更深入地了解图像处理和多媒体操作。 使用OpenCV库,你可以轻松地读取视频文件࿰…...
虹科方案 | 汽车总线协议转换解决方案(二)
上期说到,虹科的PCAN-LIN网关在CAN、LIN总线转换方面有显著的作用,尤其是为BMS电池通信的测试提供了优秀的解决方案。假如您感兴趣,可以点击文末相关链接进行回顾! 而今天,虹科将继续给大家带来Router系列在各个领域的…...
[Android] 通过JNI 让 JAVA 调用 android native 接口
前言: JNI (java native interface) 是一个库,可以让 java 代码和其他语言互动,比如 java 通过 JNI 调用融合了 jni库的 c/c 代码,注意,这里要求 c/c代码中必须通过链接 jni 库并按照 JNI 规范定义一套可供 JAVA 调用…...
MySQL高可用MHA
目录 前言 一、概述 二、配置免密、组从复制 三、MHA配置 四、测试 总结 前言 MySQL高可用管理工具(MHA,Master High Availability)是一个用于自动管理MySQL主从复制的工具,它可以提供高可用性和自动故障转移。MHA由原版的MHA工具…...
conda相比python好处
Conda 作为 Python 的环境和包管理工具,相比原生 Python 生态(如 pip 虚拟环境)有许多独特优势,尤其在多项目管理、依赖处理和跨平台兼容性等方面表现更优。以下是 Conda 的核心好处: 一、一站式环境管理:…...
docker详细操作--未完待续
docker介绍 docker官网: Docker:加速容器应用程序开发 harbor官网:Harbor - Harbor 中文 使用docker加速器: Docker镜像极速下载服务 - 毫秒镜像 是什么 Docker 是一种开源的容器化平台,用于将应用程序及其依赖项(如库、运行时环…...
基于FPGA的PID算法学习———实现PID比例控制算法
基于FPGA的PID算法学习 前言一、PID算法分析二、PID仿真分析1. PID代码2.PI代码3.P代码4.顶层5.测试文件6.仿真波形 总结 前言 学习内容:参考网站: PID算法控制 PID即:Proportional(比例)、Integral(积分&…...
k8s从入门到放弃之Ingress七层负载
k8s从入门到放弃之Ingress七层负载 在Kubernetes(简称K8s)中,Ingress是一个API对象,它允许你定义如何从集群外部访问集群内部的服务。Ingress可以提供负载均衡、SSL终结和基于名称的虚拟主机等功能。通过Ingress,你可…...
【Linux】C语言执行shell指令
在C语言中执行Shell指令 在C语言中,有几种方法可以执行Shell指令: 1. 使用system()函数 这是最简单的方法,包含在stdlib.h头文件中: #include <stdlib.h>int main() {system("ls -l"); // 执行ls -l命令retu…...
高危文件识别的常用算法:原理、应用与企业场景
高危文件识别的常用算法:原理、应用与企业场景 高危文件识别旨在检测可能导致安全威胁的文件,如包含恶意代码、敏感数据或欺诈内容的文档,在企业协同办公环境中(如Teams、Google Workspace)尤为重要。结合大模型技术&…...
USB Over IP专用硬件的5个特点
USB over IP技术通过将USB协议数据封装在标准TCP/IP网络数据包中,从根本上改变了USB连接。这允许客户端通过局域网或广域网远程访问和控制物理连接到服务器的USB设备(如专用硬件设备),从而消除了直接物理连接的需要。USB over IP的…...
CSS设置元素的宽度根据其内容自动调整
width: fit-content 是 CSS 中的一个属性值,用于设置元素的宽度根据其内容自动调整,确保宽度刚好容纳内容而不会超出。 效果对比 默认情况(width: auto): 块级元素(如 <div>)会占满父容器…...
HarmonyOS运动开发:如何用mpchart绘制运动配速图表
##鸿蒙核心技术##运动开发##Sensor Service Kit(传感器服务)# 前言 在运动类应用中,运动数据的可视化是提升用户体验的重要环节。通过直观的图表展示运动过程中的关键数据,如配速、距离、卡路里消耗等,用户可以更清晰…...
虚拟电厂发展三大趋势:市场化、技术主导、车网互联
市场化:从政策驱动到多元盈利 政策全面赋能 2025年4月,国家发改委、能源局发布《关于加快推进虚拟电厂发展的指导意见》,首次明确虚拟电厂为“独立市场主体”,提出硬性目标:2027年全国调节能力≥2000万千瓦࿰…...
