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 并…...

观成科技:隐蔽隧道工具Ligolo-ng加密流量分析
1.工具介绍 Ligolo-ng是一款由go编写的高效隧道工具,该工具基于TUN接口实现其功能,利用反向TCP/TLS连接建立一条隐蔽的通信信道,支持使用Let’s Encrypt自动生成证书。Ligolo-ng的通信隐蔽性体现在其支持多种连接方式,适应复杂网…...

测试微信模版消息推送
进入“开发接口管理”--“公众平台测试账号”,无需申请公众账号、可在测试账号中体验并测试微信公众平台所有高级接口。 获取access_token: 自定义模版消息: 关注测试号:扫二维码关注测试号。 发送模版消息: import requests da…...

React第五十七节 Router中RouterProvider使用详解及注意事项
前言 在 React Router v6.4 中,RouterProvider 是一个核心组件,用于提供基于数据路由(data routers)的新型路由方案。 它替代了传统的 <BrowserRouter>,支持更强大的数据加载和操作功能(如 loader 和…...

2025年能源电力系统与流体力学国际会议 (EPSFD 2025)
2025年能源电力系统与流体力学国际会议(EPSFD 2025)将于本年度在美丽的杭州盛大召开。作为全球能源、电力系统以及流体力学领域的顶级盛会,EPSFD 2025旨在为来自世界各地的科学家、工程师和研究人员提供一个展示最新研究成果、分享实践经验及…...

中南大学无人机智能体的全面评估!BEDI:用于评估无人机上具身智能体的综合性基准测试
作者:Mingning Guo, Mengwei Wu, Jiarun He, Shaoxian Li, Haifeng Li, Chao Tao单位:中南大学地球科学与信息物理学院论文标题:BEDI: A Comprehensive Benchmark for Evaluating Embodied Agents on UAVs论文链接:https://arxiv.…...

遍历 Map 类型集合的方法汇总
1 方法一 先用方法 keySet() 获取集合中的所有键。再通过 gey(key) 方法用对应键获取值 import java.util.HashMap; import java.util.Set;public class Test {public static void main(String[] args) {HashMap hashMap new HashMap();hashMap.put("语文",99);has…...
MVC 数据库
MVC 数据库 引言 在软件开发领域,Model-View-Controller(MVC)是一种流行的软件架构模式,它将应用程序分为三个核心组件:模型(Model)、视图(View)和控制器(Controller)。这种模式有助于提高代码的可维护性和可扩展性。本文将深入探讨MVC架构与数据库之间的关系,以…...
镜像里切换为普通用户
如果你登录远程虚拟机默认就是 root 用户,但你不希望用 root 权限运行 ns-3(这是对的,ns3 工具会拒绝 root),你可以按以下方法创建一个 非 root 用户账号 并切换到它运行 ns-3。 一次性解决方案:创建非 roo…...
python如何将word的doc另存为docx
将 DOCX 文件另存为 DOCX 格式(Python 实现) 在 Python 中,你可以使用 python-docx 库来操作 Word 文档。不过需要注意的是,.doc 是旧的 Word 格式,而 .docx 是新的基于 XML 的格式。python-docx 只能处理 .docx 格式…...

HashMap中的put方法执行流程(流程图)
1 put操作整体流程 HashMap 的 put 操作是其最核心的功能之一。在 JDK 1.8 及以后版本中,其主要逻辑封装在 putVal 这个内部方法中。整个过程大致如下: 初始判断与哈希计算: 首先,putVal 方法会检查当前的 table(也就…...