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

kafka consumer客户端消费逻辑解析

kafka consumer客户端消费逻辑解析

    • 一、主要步骤
    • 二、提交策略
      • 【步骤2代码解析】
      • 【提交策略总结】
    • 三、拉取策略
    • 四、消费策略
      • 【代码解析】
      • 【消费策略总结】

一、主要步骤

在这里插入图片描述
这是kafka客户端拉取消息的入口,有4个主要部分

1、启动后的准备
consumer线程启动后,如果非自动提交模式,构建worker线程放入worker线程池,供后续消费消息使用

2、运行期逻辑循环——提交策略

3、运行期逻辑循环——消息拉取

4、运行期逻辑循环——消息消费

二、提交策略

【步骤2代码解析】

2.1 在拉取消息之前,如果非自动提交,进行提交判定:
需要提交的消息(ConsumerRecord)会维护在acks——本地已处理待提交消息队列(一个linkedBlockingQueue)中,这里会把acks里所有消息拿出来进行循环处理。
在这里插入图片描述
(2-1-1) ack策略-立即提交:说明在配置了手动提交-立即提交的ack策略时,提交动作是每次消息拉取前,worker线程已处理完的消息的offset,挨个put进本地的partition和offset的映射(metadata中一个map)。由于是循环处理worker队列,而消息是乱序存放的,所以put之前判断offset大于现有offset才会执行,确保低offset不覆盖高offset。
put后会直接进行网络请求提交到broker中。(由于在循环中,这里的请求会发生多次?没细看)在这里插入图片描述在这里插入图片描述
在这里插入图片描述

(2-1-2) ack策略-非立即提交:
和2-1-1一样,都会维护分区最高位移映射。在这里插入图片描述
接下来会根据具体的提交规则配置来判定是否提交,
1、未提交数:未提交数 >= 配置
2、提交时间间隔:上次提交 - 当前时间 >= 配置
3、未提交数或提交时间间隔:1或2任意满足

在这里插入图片描述

【提交策略总结】

一、提交模式
1、自动提交:拉取消息后立即提交
2、手动(非自动)提交:
2-1、拉取消费前执行一次提交判定

二、提交判定
1、立即提交:无需判定
2、非立即提交:根据配置的规则判定
2-1、满足提交时间间隔可提交
2-2、满足未提交数计算可提交
2-3、满足2-1或2-2可提交

三、拉取策略

在这里插入图片描述
会循环尝试拉取消息直到超时(最大尝试拉取时长)

四、消费策略

【代码解析】

从待消费消息列表拉取消息,会循环尝试拉取消息直到超时(最大尝试拉取时长),用当前线程或分配给worker线程
从待消费消息列表拉取消息,用当前线程或分配给worker线程。worker线程和拉取线程1:1(worker线程为什么不是多个?)
在这里插入图片描述

自动提交,用当前线程直接进行消费(可能阻塞消息拉取和位移提交)
在这里插入图片描述
手动提交,加入待处理消息队列(一个linkedBlockingQueue),等待消费线程拉取并消费
在这里插入图片描述
从待处理消息队列拉取消息,循环进行消费,消费后加入已处理待提交队列
在这里插入图片描述

【消费策略总结】

【消费主体】
1、自动提交:consume线程自我消费(会阻塞消息拉取和位移提交)
2、手动提交:worker线程异步消费

【消费模式】
循环消费

相关文章:

kafka consumer客户端消费逻辑解析

kafka consumer客户端消费逻辑解析 一、主要步骤二、提交策略【步骤2代码解析】【提交策略总结】 三、拉取策略四、消费策略【代码解析】【消费策略总结】 一、主要步骤 这是kafka客户端拉取消息的入口,有4个主要部分 1、启动后的准备 consumer线程启动后&#xff…...

打印机出现多个副本无法删除

打印机出现多个副本很烦人,尤其是在打印机在局域网内被多个主机共享的时候,一旦出现新的副本,原有副本全都变成脱机状态,其他电脑连接的共享打印机是原来的副本,所以要重新设置打印机共享,很烦人。 想要删…...

FlinkSQL 开发经验分享

作者:汤包 最近做了几个实时数据开发需求,也不可避免地在使用 Flink 的过程中遇到了一些问题,比如数据倾斜导致的反压、interval join、开窗导致的水位线失效等问题,通过思考并解决这些问题,加深了我对 Flink 原理与机…...

JVM原理(十二):JVM虚拟机类加载过程

一个类型从被加载到虚拟机内存中开始,到卸载为止,它的整个生命周期将会经过 加载、验证、准备、解析、初始化、使用、卸载七个阶段。其中 验证、准备、解析三个部分统称为 连接 1. 加载 加载是整个类加载的一个过程。在加载阶段,Java虚拟机…...

Apipost接口测试工具的原理及应用详解(三)

本系列文章简介: 随着软件行业的快速发展,API(应用程序编程接口)作为不同软件组件之间通信的桥梁,其重要性日益凸显。API的质量直接关系到软件系统的稳定性、性能和用户体验。因此,对API进行严格的测试成为软件开发过程中不可或缺的一环。在众多API测试工具中,Apipost凭…...

unity里鼠标位置是否在物体上。

1. 使用Raycast 如果你的图片是在UI Canvas上,可以使用Raycast来检测鼠标点击是否在图片上。 using UnityEngine; using UnityEngine.EventSystems; using UnityEngine.UI; public class ImageClickChecker : MonoBehaviour { public Image targetImage; voi…...

Java知识点大纲

文章目录 第一阶段:JavaSE1、面向对象编程(基础)1)面向过程和面向对象区别2)类和对象的概述3)类的属性和方法4)创建对象内存分析5)构造方法(Construtor)及其重载6)对象类型的参数传递7)this关键字详解8)static关键字详解9)局部代码块、构造代码块和静态代码块10)pac…...

【Kafka】记录一次Kafka消费者重复消费问题

文章目录 现象业务背景排查过程Push与Pull 现象 用户反馈消费者出现消息积压,并且通过日志看,一直重复消费,且没有报错日志。 业务背景 用户的消费者是一个将文件做Embedding的任务,(由于AI技术的兴起,大…...

Android使用http加载自建服务器静态网页

最终效果如下图,成功加载了电脑端的静态网页内容,这是一个xml文件。 电脑端搭建http服务器 使用“Apache Http Server”,下载地址是:https://httpd.apache.org/download.cgi。 安装启动步骤,参考:Apach…...

python解耦重构,提高程序维护性

一、重构思想 思路来源 java spring设计模式学习,强调低耦合的思想,通过解耦来提高程序的可维护性。 二、代码重构 解决方案 通过单独配置文件来控制变量的改变。 spring的话可以读取xml或者是springboot 读取application.properties 来获取变量值。…...

深入解析 Laravel 事件系统:架构、实现与应用

Laravel 的事件系统是框架中一个强大且灵活的功能,它允许开发者在应用程序中定义和使用自定义事件和监听器。这个系统基于观察者模式,使得代码解耦和可维护性大大提高。在本文中,我们将深入探讨 Laravel 事件系统的工作原理、如何实现自定义事…...

视频怎么制作gif动态图片?GIF制作方法分享

视频怎么制作gif动态图片?视频制作GIF动态图片,不仅保留了视频的生动瞬间,还赋予了图像循环播放的魔力。这一技能不仅让创意表达更加丰富多彩,还极大地提升了视觉传播的效率和趣味性。在快节奏的数字时代,GIF动图以其小…...

js 使用 lodash-es 检测某个值是否是函数

import { isFunction } from lodash-eslet isA isFunction(() > {}) console.log(isA) //true https://www.lodashjs.com/docs/lodash.isFunction#_isfunctionvalue https://lodash.com/docs/4.17.15#isFunction 人工智能学习网站 https://chat.xutongbao.top...

[go-zero] goctl 生成api和rpc

文章目录 1.goctl 概述2.go-zero 需要安装的组件3.生成 api4.生成 rpc 1.goctl 概述 goctl支持多种rpc,较为流行的是google开源的grpc,这里主要介绍goctl rpc protoc的代码生成与使用。protoc是grpc的命令,作用是将proto buffer文件转化为相…...

docker -run hello-world超时

主要原因就是尝试拉取库的时候没有从阿里云镜像里拉&#xff0c;所以设置一下就好了 这里使用的是ubuntu系统&#xff08;命令行下逐行敲就行了&#xff09; sudo mkdir -p /etc/docker sudo tee /etc/docker/daemon.json <<-EOF {"registry-mirrors": [&quo…...

拓扑学习系列(8)黎曼度量曲面与高斯曲率

黎曼度量 黎曼度量是黎曼几何中的一个重要概念&#xff0c;它用来描述流形上的切向量之间的长度和角度。黎曼度量赋予了流形一个内积结构&#xff0c;使得我们可以定义切向量的长度、夹角和内积&#xff0c;从而引入了度量空间的概念。让我更详细地解释一下黎曼度量&#xff1…...

汽车IVI中控开发入门及进阶(三十四):i.MX linux BSP

开发板: 汽车IVI中控开发入门及进阶(三十三):i.MX linux开发之开发板-CSDN博客 linux 开发项目: 汽车IVI中控开发入门及进阶(三十二):i.MX linux开发之Yocto-CSDN博客 前言: 有了开发板,linux BSP编译项目yocto,接下来就可以在i.MX平台上构建和安装i.MX Linux …...

【Python机器学习】算法链与管道——构建管道

目录 1、首先&#xff0c;我们构建一个由步骤列表组成的管道对象。 2、向任何其他scikit-learn估计器一样来拟合这个管道 3、调用pipe.score 我们来看下如何使用Pipeline类来表示在使用MinMaxScaler缩放数据后&#xff0c;再训练一个SVM的工作流程&#xff08;暂时不用网格搜…...

Postman 高级用法学习

Postman 高级用法 Postman 是一款强大的 API 调试和开发工具&#xff0c;广泛应用于 API 开发、测试、调试和自动化流程中。除了基本的 API 请求发送和响应查看功能&#xff0c;Postman 还提供了许多高级功能。以下是详细的讲解&#xff0c;包括具体示例和操作步骤。 一、环境…...

从新手到高手:Scala函数式编程完全指南,Scala 访问修饰符(6)

1、Scala 访问修饰符 Scala 访问修饰符基本和Java的一样&#xff0c;分别有&#xff1a;private&#xff0c;protected&#xff0c;public。 如果没有指定访问修饰符&#xff0c;默认情况下&#xff0c;Scala 对象的访问级别都是 public。 Scala 中的 private 限定符&#xff…...

连锁超市冷库节能解决方案:如何实现超市降本增效

在连锁超市冷库运营中&#xff0c;高能耗、设备损耗快、人工管理低效等问题长期困扰企业。御控冷库节能解决方案通过智能控制化霜、按需化霜、实时监控、故障诊断、自动预警、远程控制开关六大核心技术&#xff0c;实现年省电费15%-60%&#xff0c;且不改动原有装备、安装快捷、…...

WordPress插件:AI多语言写作与智能配图、免费AI模型、SEO文章生成

厌倦手动写WordPress文章&#xff1f;AI自动生成&#xff0c;效率提升10倍&#xff01; 支持多语言、自动配图、定时发布&#xff0c;让内容创作更轻松&#xff01; AI内容生成 → 不想每天写文章&#xff1f;AI一键生成高质量内容&#xff01;多语言支持 → 跨境电商必备&am…...

项目部署到Linux上时遇到的错误(Redis,MySQL,无法正确连接,地址占用问题)

Redis无法正确连接 在运行jar包时出现了这样的错误 查询得知问题核心在于Redis连接失败&#xff0c;具体原因是客户端发送了密码认证请求&#xff0c;但Redis服务器未设置密码 1.为Redis设置密码&#xff08;匹配客户端配置&#xff09; 步骤&#xff1a; 1&#xff09;.修…...

C#学习第29天:表达式树(Expression Trees)

目录 什么是表达式树&#xff1f; 核心概念 1.表达式树的构建 2. 表达式树与Lambda表达式 3.解析和访问表达式树 4.动态条件查询 表达式树的优势 1.动态构建查询 2.LINQ 提供程序支持&#xff1a; 3.性能优化 4.元数据处理 5.代码转换和重写 适用场景 代码复杂性…...

适应性Java用于现代 API:REST、GraphQL 和事件驱动

在快速发展的软件开发领域&#xff0c;REST、GraphQL 和事件驱动架构等新的 API 标准对于构建可扩展、高效的系统至关重要。Java 在现代 API 方面以其在企业应用中的稳定性而闻名&#xff0c;不断适应这些现代范式的需求。随着不断发展的生态系统&#xff0c;Java 在现代 API 方…...

人工智能 - 在Dify、Coze、n8n、FastGPT和RAGFlow之间做出技术选型

在Dify、Coze、n8n、FastGPT和RAGFlow之间做出技术选型。这些平台各有侧重&#xff0c;适用场景差异显著。下面我将从核心功能定位、典型应用场景、真实体验痛点、选型决策关键点进行拆解&#xff0c;并提供具体场景下的推荐方案。 一、核心功能定位速览 平台核心定位技术栈亮…...

32单片机——基本定时器

STM32F103有众多的定时器&#xff0c;其中包括2个基本定时器&#xff08;TIM6和TIM7&#xff09;、4个通用定时器&#xff08;TIM2~TIM5&#xff09;、2个高级控制定时器&#xff08;TIM1和TIM8&#xff09;&#xff0c;这些定时器彼此完全独立&#xff0c;不共享任何资源 1、定…...

FOPLP vs CoWoS

以下是 FOPLP&#xff08;Fan-out panel-level packaging 扇出型面板级封装&#xff09;与 CoWoS&#xff08;Chip on Wafer on Substrate&#xff09;两种先进封装技术的详细对比分析&#xff0c;涵盖技术原理、性能、成本、应用场景及市场趋势等维度&#xff1a; 一、技术原…...

嵌入式面试常问问题

以下内容面向嵌入式/系统方向的初学者与面试备考者,全面梳理了以下几大板块,并在每个板块末尾列出常见的面试问答思路,帮助你既能夯实基础,又能应对面试挑战。 一、TCP/IP 协议 1.1 TCP/IP 五层模型概述 链路层(Link Layer) 包括网卡驱动、以太网、Wi‑Fi、PPP 等。负责…...

Python异步编程:深入理解协程的原理与实践指南

&#x1f49d;&#x1f49d;&#x1f49d;欢迎莅临我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 持续学习&#xff0c;不断…...