当前位置: 首页 > news >正文

Kafka 入门到起飞系列 - 消费者组管理、位移管理

消费者组 - Consumer Group

上文我们已经讲过消费者组了,我们知道消费组的存在可以保证一个主题下一个分区的消息只会被组内一个消费者消费,从而避免了消息的重复消费

什么是消费组 - Consumer Group?

消费者组是Kafka 提供的可扩展且具有容错性的消费者机制

消费组的三大特性

  1. 消费组有一个或多个消费者,消费者可以是一个服务、一个进程、一个线程
  2. 消费组具有一个唯一标识字符串,group.id
  3. 消费组订阅的主题每个分区只能分配给一个消费者

怎么理解呢,其实就是让Kafka的客户端(此时指消费者)具有弹性,弹性伸缩,可扩展,具有容错性
比如我们有一个主题,主题有5个分区,我们有一个消费组,组内有3个消费者组成消费者集群,如下图
在这里插入图片描述
比如此时消费者消费能力不够了,我们可以在线添加一个消费者到消费组,提升消费能力,当然不管增加消费者还是消费者下线,都需要重新分配分区与消费者的关系(Rebalance)如下图
在这里插入图片描述

消费组的存在就是说: 以消费组的名义,去消费主题的消息,当然最终消费组里的消费者有多少取决于主题有多少个分区
记录偏移量,多于分区个数的消费者会分配不到分区而造成空闲;同时也是以消费组的名义,在Kafka里记录消费消息的偏移量


既然讲到偏移量,那我们来看看偏移量(位移)是怎么管理的?

位移管理

首先位移就是偏移量(offset),就是消息的编号,Kafka内置了一个主题_consumer_offset来记录分区消费情况(该分区消费到第几号消息了),那么为什么说偏移量是由消费者自己管理的呢?因为是消费者自己决定消费分区里第几号消息,然后把这个偏移量信息定期提交到Kafka去,Kafka再将偏移量信息记录到_consumer_offset主题去,所以其实消费者本身也是一个生产者

消费者提交偏移量的两种方式

  • 自动提交,Kafka默认自动提交,可以通过参数调整 enable.auto.commit = false即关闭自动提交
  • 手动提交

Kafka 是如何记录偏移量信息的呢?

消费者是在消费消息的过程中定期提交偏移量的
Kafka会将group消费情况保存起来,做成一个offset map,如下图:

key: groupId + Topic + partition
value:  offset

绿色代表已经消费过了,红色代表未被消费消息,箭头代表偏移量,下一次要消费的消息
在这里插入图片描述
在文件里实际存储形式如下:
在这里插入图片描述

以消费组的名义记录偏移量信息
这里我们发现记录偏移量时没有消费者什么事,这也是Kafka设计的精妙之处,
任意消费者分配了分区后只要以消费组的名义都能获取分区数据,
当消费者伸缩重新分配分区后,依然可以继续消费,
因为偏移量没有跟消费者绑定,是跟消费组绑定的,这也是一种容错机制类似于Java里面多态的一种思想

相关文章:

Kafka 入门到起飞系列 - 消费者组管理、位移管理

消费者组 - Consumer Group 上文我们已经讲过消费者组了,我们知道消费组的存在可以保证一个主题下一个分区的消息只会被组内一个消费者消费,从而避免了消息的重复消费 什么是消费组 - Consumer Group? 消费者组是Kafka 提供的可扩展且具有容…...

SpringBoot——数据层三组件之间的关系

简单介绍 在之前的文章中,我们介绍了一下SpringBoot中内置的几种数据层的解决方案,在数据层由三部分组成,分别是数据库,持久化技术以及数据源,但是我今天写着写着,突然就想不起来这三部分到底是干什么的了…...

LeetCode647.Palindromic-Substrings<回文子串>

题目: 思路: 错误代码:(缺少部分判断) 使用的是寻找回文子串的方法。以一个点为中心向两边扫描。但是有一点小问题。 因为回文子串是分奇偶的,所以需要两种判断方式。 看了下答案后发现我的代码距离答案一…...

React的hooks---useContext

Context 提供了一个无需为每层组件手动添加 props ,就能在组件树间进行数据传递的方法,useContext 用于函数组件中订阅上层 context 的变更,可以获取上层 context 传递的 value prop 值 useContext 接收一个 context 对象(React.…...

【Terraform学习】TerraformCloud入门介绍(快速入门)

TerraformCloud入门介绍 什么是 TerraformCloud? Terraform Cloud是Hashicorp Terraform的SaaS版本。 免费版功能 免费版功能包括版本控制集成、远程计划和实施远程计划和实施、通知及webhook、全http API驱动、状态管理、模拟计划、私有化模块注册器以及全HTTP界…...

linux实现运行java分包部署

1.打好包之后 找到bin文件夹下的 startup.sh文件 2.cd 进入bin文件夹下 3.执行 sh startup.sh 运行命令 4.如果出现此错误 是Windows和Linux的.sh脚本文件格式不同,如果在脚本中有空行,脚本是在Windows下进行编辑之后上传到linux上去执行的话&#xff0c…...

数据安全之全景图系列——数据分类分级落地实践

1、数据分类分级现状 我们正处于一个数据爆炸式增长的时代,随着产业数字化转型升级的推进,数据已被国家层面纳入生产要素,并且成为企业、社会和国家层面重要的战略资源。数据分类分级管理不仅是加强数据交换共享、提升数据资源价值的前提条件…...

C++实现MySQL数据库连接池

C实现MySQL数据库连接池 涉及技术 MySQL数据库编程、单例模式、STL容器、C11多线程(线程互斥、线程互斥、线程同步通信和unique_lock)、智能指针shared_ptr、lambda表达式、生产者-消费者线程模型。 项目背景 为了提升MySQL数据库(基于C/…...

day4 驱动开发 c语言学习

不利用系统提供的register_chrdev&#xff0c;自己实现字符设备的注册 底层代码 led.c #include <linux/init.h> #include <linux/module.h> #include <linux/fs.h> #include <linux/uaccess.h> #include <linux/io.h> #include "head.h…...

history命令:显示命令执行时间

1.修改配置文件 vim /etc/profile 添加内容 export HISTTIMEFORMAT"%Y-%m-%d %H:%M:%S " ​ #注意&#xff1a;在末尾的“引号”与“S”之间&#xff0c;加入一位空格&#xff0c;将日期时间和历史命令用空格相隔开来。 你也可以换一种清晰的形式&#xff0c;效果…...

Django接口返回JSON格式数据报文

目录 遇到问题 Django返回json结构报文 不可行方式python json 可行方式JsonResponseQuerySet.values()。 python的两个web框架&#xff0c;flask和django&#xff0c;两者都具有view 模板的章节&#xff0c;但是当前开发一个应用&#xff0c;大部分采用前后端分离的合作方式。…...

OBS 迁移--华为云

一、创建迁移i任务 1. 登录管理控制台。 2. 单击管理控制台左上角的 在下拉框中选择区域。 3. 单击“ 服务列表 ”&#xff0c;选择“ 迁移 > 对象存储迁移服务 OMS ”&#xff0c;进入“ 对象存储迁移服务 ”页面。 4. 单击页面右上角“ 创建迁移任务 ”。 5. 仔细阅读…...

【Docker consul的容器服务更新与发现】

文章目录 一、Consul 的简介&#xff08;1&#xff09;什么是服务注册与发现&#xff08;2&#xff09;什么是consul 二、consul 部署1、consul服务器1. 建立 Consul 服务2. 查看集群信息3. 通过 http api 获取集群信息 2、registrator服务器1. 安装 Gliderlabs/Registrator2. …...

MFC第二十天 数值型关联变量 和单选按钮与复选框的开发应用

文章目录 数值型关联变量数值型关联变量的种类介绍 单选按钮与复选框单选按钮的组内选择原理解析单选按钮和复选框以及应用数值型关联变量的开发CMainDlg.cppCInputDlg.hCInputDlg.cpp 附录 数值型关联变量 数值型关联变量的种类介绍 1、 数值型关联变量&#xff1a; a)控件型…...

服务器 Docker Alist挂载到本地磁盘(Mac版)夸克网盘

1.服务器下载alist 默认有docker环境 docker pull xhofe/alist2.生成容器 -v /home/alist:/opt/alist/data 这段意思是alist中的数据映射到docker 主机的文件夹&#xff0c;/home/alist就是我主机的文件夹&#xff0c;这个文件夹必须先创建 docker run -d --restartalways…...

EMP-SSL: TOWARDS SELF-SUPERVISED LEARNING IN ONETRAINING EPOCH

Recently, self-supervised learning (SSL) has achieved tremendous success in learning image representation. Despite the empirical success, most self-supervised learning methods are rather “inefficient” learners, typically taking hundreds of training epoch…...

注解和反射01--什么是注解

注解 什么是注解内置注解元注解自定义注解 什么是注解 1、注解是从JDK5.0开始引入的新技术 2、注解的作用 &#xff08;1&#xff09;不是程序本身&#xff0c;可以对程序做出解释&#xff08;和注释相同&#xff09; &#xff08;2&#xff09;可以被其他程序读取&#xff0c…...

虚拟机 RHEL8 安装 MySQL 8.0.34

目录 安装步骤一、清除所有残留的旧MySQL二、安装MySQL 报错问题1. 提示未找到匹配的参数&#xff1a; mysql-community-server2. 公钥问题 安装步骤 一、清除所有残留的旧MySQL 1. 关闭MySQL [rootlocalhost /]# service mysqld stop Redirecting to /bin/systemctl stop …...

kafka 总结宝典

...

跨平台力量:探索C++Qt框架的未来前景

卓越的跨平台支持&#xff1a;CQt可以在多个平台上运行&#xff0c;包括Windows、Mac、Linux、Android和iOS等。这使得开发者能够使用同一份代码构建跨平台的应用程序&#xff0c;从而显著降低了开发成本和时间投入。 丰富的类库和工具&#xff1a;CQt提供了广泛的类库和工具&…...

基于长短期神经网络LSTM的位移监测,基于长短期神经网络的位移预测,LSTM的详细原理

目录 背影 摘要 LSTM的基本定义 LSTM实现的步骤 基于长短期神经网络LSTM的位移监测 完整代码: https://download.csdn.net/download/abc991835105/88098131 效果图 结果分析 展望 参考论文 背影 路径追踪预测,对实现自动飞行驾驶拥有重要意义,长短期神经网络是一种改进党的…...

ChatGPT漫谈(二)

ChatGPT“脱胎”于OpenAI在2020年发布的GPT-3,任何外行都可以使用GPT-3,在几分钟内提供示例,并获得所需的文本输出。GPT-3被认为是当时最强大的语言模型,但现在,ChatGPT模型似乎更强大。ChatGPT能进行天马行空的长对话,可以回答问题,它具备了类人的逻辑、思考与沟通的能…...

【LangChain】检索器之MultiQueryRetriever

MultiQueryRetriever 概要内容总结 概要 基于距离的向量数据库检索在高维空间中嵌入查询&#xff0c;并根据“距离”查找相似的嵌入文档。 但是&#xff0c;如果查询措辞发生细微变化&#xff0c;或者嵌入不能很好地捕获数据的语义&#xff0c;检索可能会产生不同的结果。有时…...

教师ChatGPT的23种用法

火爆全网的ChatGPT&#xff0c;作为教师应该如何正确使用&#xff1f;本文梳理了教师ChatGPT的23种用法&#xff0c;一起来看看吧&#xff01; 1、回答问题 ChatGPT可用于实时回答问题&#xff0c;使其成为需要快速获取信息的学生的有用工具。 从这个意义上说&#xff0c;Cha…...

【libevent】http客户端1:转存http下载的数据

read_http_input // // HTTP endpoint: GET /rpc/1 (list methods) or POST /rpc/1 (execute RPC) // // JSON-RPC API endpoint. Handles all JSON-RPC method calls. // static void rpc_jsonrpc(evhttp_request *req, void *opaque) {RpcApiInfo *ap =...

Pytorch学习笔记 | 数据类型 | mnist数据集

数据类型 python中数据类型和pytorch中的对应关系 注意:pytorch是没有没有string类型的 例1:创建一个3行4列的随机数数组,符合均值为0,方差为1的正态分布 import torch a=torch.Tensor(3,4) a Out[17]: tensor([[0....

Linux虚拟机(lvm)报Unmount and run xfs_repair

问题 linux系统没有正常关机&#xff0c;今天启动虚拟机无法进入系统&#xff0c;提示metadata corruption deleted at xxxx&#xff1b; Unmount and run xfs_repair 分析 主机异常掉电后里面的虚拟机无法启动&#xff0c;主要是损坏的分区 解决 看出来应该是dm-0分区损坏…...

【ESP32】Espressif-IDE及ESP-IDF安装

一、下载Espressif-IDE 2.10.0 with ESP-IDF v5.0.2 1.打开ESP-IDF 编程指南 2.点击快速入门–>安装–>手动安装–>Windows Installer–>Windows Installer Download 3.点击下载Espressif-IDE 2.10.0 with ESP-IDF v5.0.2 二、安装Espressif-IDE 2.10.0 wit…...

基于vue3实现画布操作的撤销与重做

基于vue3实现画布操作的撤销与重做 前言 vue3项目中实现在canvas画布上实现画节点和连线功能&#xff0c;要求可以撤销重做 思路 canvasBox 画板数据是存放在对象里面&#xff1b; snapshots存放操作记录&#xff1b; curIndex表示当前操作索引的下标&#xff1b; maxLimit表…...

php 抽象工厂模式

1&#xff0c;抽象工厂&#xff08;Abstract Factory&#xff09;模式&#xff0c;是创建设计模式的一种&#xff0c;它创建一系列相关的对象&#xff0c;而不必指定具体的类。该模式为一个产品族提供了统一的创建接口。当需要这个产品族的某一系列的时候&#xff0c;可以为此系…...