kafka 超详细的消息订阅与消息消费几种方式
kafka 消息订阅与消息消费几种方式
本文主要内容
-
消费者订阅几种方式
-
订阅多个主题
-
按正则表达式订阅
-
-
消息消费几种方式
-
按分区消费
-
按主题消费
-
不区分
-
“笔者建议一开始学习Kafka最好不要用SpringBoot 集成方式,因为SpringBoot推崇用注解方式,比如
@KafkaListener
等,就可以直接消费,这样不能直接接触kafka-client一些api, 且SpringBoot 给我们提供了很多默认配置,我们几乎零配置
也可以使用,实际上kafka很多配置很重要的,不容忽视。
消费者订阅几种方式
KafkaConsumer
给我们提供了几种订阅消息方式,我们可以订阅多个消息。示例代码如下
kafkaConsumer.subscribe(Arrays.asList("topicA","topicB"));kafkaConsumer.subscribe(Pattern.compile("topic-*"));kafkaConsumer.assign(Arrays.asList(new TopicPartition("topicA",0)));
订阅多个主题
void subscribe(Collection<String> topics)
对应上面第一行代码,这是最常见的订阅方式
按正则表达式订阅
void subscribe(Pattern pattern)
符合正则的主题都会被消费
“有人创建了新的主题,并且与正则匹配,消费者也可以消费到
这种方式需要能对多种消息处理,对于一些能通用处理,不感知具体业务数据的场景比较合适。比如B系统需要同步A系统数据,我们按正则订阅,当A系统有新的数据需要同步,这是只需要A发满足条件正则的消息,B系统无需任何改动。
订阅指定分区
void assign(Collection<TopicPartition> partitions);
正常业务不会使用,如果订阅的分区不存在,会报错。一些特殊场景,比如需要精确控制消费者消费消息,自定义分区分配策略时 可能会用到assign 方法
消息消费
“kafka 采用客户端
拉取模式
进行消息消费
poll() 返回所订阅的主题上一组消息ConsumerRecords
,我们可以对消息进行按主题、按分区进行处理,当然可以统一处理,不分主题和分区
ConsumerRecords<String,String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
不区分主题、分区
for(ConsumerRecord<String,String> record : records){// 处理消息
}
按partition 处理
Set<TopicPartition> topicPartitions = records.partitions();
for(TopicPartition topicPartition : topicPartitions){List<ConsumerRecord<String, String>> tpRecords = records.records(topicPartition);
}
按主题
Iterable<ConsumerRecord<String,String>> iter = records.records(topic);
思考
kafka 给我们提供了灵活的消息订阅以及消息消费方式,我们需要根据实际业务场景选择。无论哪种场景都离不开 主题
、分区
,最主要的是分区
,当我们选择了某种订阅方式如果主题
、分区
发生了变化 ,消息还能正常消费吗
“选择了按正则订阅消息方式, 后面创建了新的主题,该消息能被正常消费吗
“选择了指定分区订阅, 如果后面扩容了新的分区,新分区消息能消费吗?
List<PartitionInfo> partitionsFor(String topic)
能获取分区情况,如果需要按分区订阅,该方法一定用的上
“按分区维度消费消息,对于手动提交消息位移场景非常有用
“按主题分类处理消息也很常见,因为不同主题消息格式可能是不一样的,根据主题区分,很容易将不同的消息分类处理。
相关文章:
kafka 超详细的消息订阅与消息消费几种方式
kafka 消息订阅与消息消费几种方式 本文主要内容 消费者订阅几种方式 订阅多个主题 按正则表达式订阅 消息消费几种方式 按分区消费 按主题消费 不区分 “ 笔者建议一开始学习Kafka最好不要用SpringBoot 集成方式,因为SpringBoot推崇用注解方式,比如KafkaList…...

C++ 第三讲:内存管理
C 第三讲:内存管理 1.C内存分布2.内存管理方式2.1C语言内存管理方式2.2C内存管理方式2.2.1new\delete操作内置类型2.2.2new\delete操作自定义类型 3.operator new与operator delete函数4.new和delete实现原理4.1内置类型4.2自定义类型 5.定位new5.1内存池的基本了解…...
LeeCode打卡第二十九天
LeeCode打卡第二十九天 第一题:岛屿数量(LeeCode第200题): 给你一个由 1(陆地)和 0(水)组成的的二维网格,请你计算网格中岛屿的数量。岛屿总是被水包围,并且每座岛屿只…...
阿里云专业翻译api对接
最近我们一个商城项目涉及多语言切换,默认中文。用户切换语言可选英语和阿拉伯语言,前端APP和后端返回动态数据都要根据用户选择语言来展示。前端静态内容都做了三套语言,后端商品为了适用这种多语言我们也进行了改造。每一件商品名称&#x…...

基于Spring Boot的能源管理系统+建筑能耗+建筑能耗监测系统+节能监测系统+能耗监测+建筑能耗监测
介绍 建筑节能监测系统是基于计算机网络、物联网、大数据和数据可视化等多种技术融合形成的一套节能监测系统。 系统实现了对建筑电、水、热,气等能源、资源消耗情况的实时监测和预警、动态分析和评估,为用户建立了科学、系统的节能分析方法,…...

大数据新视界 --大数据大厂之 Cassandra 分布式数据库:高可用数据存储的新选择
💖💖💖亲爱的朋友们,热烈欢迎你们来到 青云交的博客!能与你们在此邂逅,我满心欢喜,深感无比荣幸。在这个瞬息万变的时代,我们每个人都在苦苦追寻一处能让心灵安然栖息的港湾。而 我的…...

ROS第五梯:ROS+VSCode+C++单步调试
解决问题:在ROS项目中进行断点调试。 第一步:创建一个ROS项目或者打开一个现有的ROS项目。 第二步:修改c_cpp_properties.json 增加一段命令: "compileCommands": "${workspaceFolder}/build/compile_commands.json"第三…...

SLA 概念和计算方法
SLA 概念和计算方法 SLA SLA:服务等级协议(简称:SLA,全称:service level agreement) 网站服务可用性的一个保证 9越多代表全年服务可用时间越长服务更可靠,停机时间越短,反之亦然…...

C++比大小游戏
目录 开头程序程序的流程图程序游玩的效果下一篇博客要说的东西 开头 大家好,我叫这是我58。 程序 #include <iostream> #include <Windows.h> using namespace std; int main() {int ir 1;char chparr[2] { 0 };int ip1 0;int ip2 0;int i 1;c…...

PCIe进阶之TL:Memory, I/O, and Configuration Request Rules TPH Rules
1 Memory, I/O, and Configuration Request Rules 下述规则适用于 Memory 请求、IO 请求和配置请求。 除了公共的 header 字段外,所有 Memory 请求、IO 请求和配置请求还包括以下字段: (1)Requester ID[15:0] 和 Tag[9:0],组成了 Transaction ID 。 (2)Last DW BE[3:0]…...

【初阶数据结构】一文讲清楚 “堆” 和 “堆排序” -- 树和二叉树(二)(内含TOP-K问题)
文章目录 前言1. 堆1.1 堆的概念1.2 堆的分类 2. 堆的实现2.1 堆的结构体设置2.2 堆的初始化2.3 堆的销毁2.4 添加数据到堆2.4.1 "向上调整"算法 2.5 从堆中删除数据2.5.1 “向下调整”算法 2.6 堆的其它各种方法接口函数 3. 堆排序3.1 堆排序的代码实现 4. TOP-K问题…...

sqli-lab靶场学习(二)——Less8-10(盲注、时间盲注)
Less8 第八关依然是先看一般状态 http://localhost/sqli-labs/Less-8/?id1 然后用单引号闭合: http://localhost/sqli-labs/Less-8/?id1 这关的问题在于报错是不显示,那没办法通过上篇文章的updatexml大法处理。对于这种情况,需要用“盲…...

Dijkstra算法和BFS算法(单源最短路径)
基于你设计的带权有向图,从某一结点出发,执行Dijkstra算法求单源最短路径。用文字描述每一轮执行的过程 文字描述:用BFS算法求单源最短路径的过程 Dijkstra 算法 BFS算法 广度优先算法...

在WordPress中最佳Elementor主题推荐:专家级指南
对于已经在WordPress和Elementor上有丰富经验的用户来说,选择功能强大且高度灵活的主题,能大大提升网站的表现和定制能力。今天,我们来介绍六款适合用户的专家级Elementor主题:Sydney、Blocksy、Rife Free、Customify、Deep和Laye…...
关于RabbitMQ消息丢失的解决方案
RabbitMQ如何保证消息的可靠性传输 一、消息丢失的原因 1. 生产者端 网络问题: 原因:生产者与RabbitMQ服务器之间的网络连接不稳定或中断,导致消息在传输过程中丢失。解决方案:确保网络连接稳定,监控网络状态&#x…...

c语言动态内存分配
前言 我们已经掌握的内存开辟⽅式有: int val 20;//在栈空间上开辟四个字节 char arr[10] {0};//在栈空间上开辟10个字节的连续空间 但是上述的开辟空间的⽅式有两个特点: • 空间开辟⼤⼩是固定的。 • 数组在申明的时候,必须指定数组的…...

零基础制作一个ST-LINK V2 附PCB文件原理图 AD格式
资料下载地址:零基础制作一个ST-LINK V2 附PCB文件原理图 AD格式 ST-LINK/V2是一款可以在线仿真以及下载STM8以及STM32的开发工具。支持所有带SWIM接口的STM8系列单片机;支持所有带JTAG / SWD接口的STM32系列单片机。 基本属性 ST-LINK/V2是ST意法半导体为评估、开…...

nginx基础篇(一)
文章目录 学习链接概图一、Nginx简介1.1 背景介绍名词解释 1.2 常见服务器对比IISTomcatApacheLighttpd其他的服务器 1.3 Nginx的优点(1)速度更快、并发更高(2)配置简单,扩展性强(3)高可靠性(4)热部署(5)成本低、BSD许可证 1.4 Nginx的功能特性及常用功能基本HTTP服…...

监控系列之-Grafana面板展示及制作
一 Grafana设置添加数据源 1、设置Grafana中文显示 最后保存退出,数据源添加完毕 2、导入node_exporter主机监控面板 此处 有外网的情况下,直接输入对应面板的ID号,然后点击加载即可;无无外网的话,则考虑使用上传仪表…...
值传递和地址传递
值传递 我们从下面这段代码开始: point(char*pt); void main(){char b[4]{m,n,o,p},*ptb;point(pt);printf("%c\n",*pt); } point(char *p){p3; }这段代码定义了一个函数 point 和一个主函数 main。 在 main 函数中,定义了一个字符数组 b 并…...

CTF show Web 红包题第六弹
提示 1.不是SQL注入 2.需要找关键源码 思路 进入页面发现是一个登录框,很难让人不联想到SQL注入,但提示都说了不是SQL注入,所以就不往这方面想了 先查看一下网页源码,发现一段JavaScript代码,有一个关键类ctfs…...

2.Vue编写一个app
1.src中重要的组成 1.1main.ts // 引入createApp用于创建应用 import { createApp } from "vue"; // 引用App根组件 import App from ./App.vue;createApp(App).mount(#app)1.2 App.vue 其中要写三种标签 <template> <!--html--> </template>…...

屋顶变身“发电站” ,中天合创屋面分布式光伏发电项目顺利并网!
5月28日,中天合创屋面分布式光伏发电项目顺利并网发电,该项目位于内蒙古自治区鄂尔多斯市乌审旗,项目利用中天合创聚乙烯、聚丙烯仓库屋面作为场地建设光伏电站,总装机容量为9.96MWp。 项目投运后,每年可节约标煤3670…...
Qt Http Server模块功能及架构
Qt Http Server 是 Qt 6.0 中引入的一个新模块,它提供了一个轻量级的 HTTP 服务器实现,主要用于构建基于 HTTP 的应用程序和服务。 功能介绍: 主要功能 HTTP服务器功能: 支持 HTTP/1.1 协议 简单的请求/响应处理模型 支持 GET…...
数据链路层的主要功能是什么
数据链路层(OSI模型第2层)的核心功能是在相邻网络节点(如交换机、主机)间提供可靠的数据帧传输服务,主要职责包括: 🔑 核心功能详解: 帧封装与解封装 封装: 将网络层下发…...

SpringBoot+uniapp 的 Champion 俱乐部微信小程序设计与实现,论文初版实现
摘要 本论文旨在设计并实现基于 SpringBoot 和 uniapp 的 Champion 俱乐部微信小程序,以满足俱乐部线上活动推广、会员管理、社交互动等需求。通过 SpringBoot 搭建后端服务,提供稳定高效的数据处理与业务逻辑支持;利用 uniapp 实现跨平台前…...

2025盘古石杯决赛【手机取证】
前言 第三届盘古石杯国际电子数据取证大赛决赛 最后一题没有解出来,实在找不到,希望有大佬教一下我。 还有就会议时间,我感觉不是图片时间,因为在电脑看到是其他时间用老会议系统开的会。 手机取证 1、分析鸿蒙手机检材&#x…...

UR 协作机器人「三剑客」:精密轻量担当(UR7e)、全能协作主力(UR12e)、重型任务专家(UR15)
UR协作机器人正以其卓越性能在现代制造业自动化中扮演重要角色。UR7e、UR12e和UR15通过创新技术和精准设计满足了不同行业的多样化需求。其中,UR15以其速度、精度及人工智能准备能力成为自动化领域的重要突破。UR7e和UR12e则在负载规格和市场定位上不断优化…...
06 Deep learning神经网络编程基础 激活函数 --吴恩达
深度学习激活函数详解 一、核心作用 引入非线性:使神经网络可学习复杂模式控制输出范围:如Sigmoid将输出限制在(0,1)梯度传递:影响反向传播的稳定性二、常见类型及数学表达 Sigmoid σ ( x ) = 1 1 +...

华为云Flexus+DeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建
华为云FlexusDeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建 前言 如今大模型其性能出色,华为云 ModelArts Studio_MaaS大模型即服务平台华为云内置了大模型,能助力我们轻松驾驭 DeepSeek-V3/R1,本文中将分享如何…...