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

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推崇用注解方式&#xff0c;比如KafkaList…...

C++ 第三讲:内存管理

C 第三讲&#xff1a;内存管理 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打卡第二十九天 第一题&#xff1a;岛屿数量&#xff08;LeeCode第200题&#xff09;: 给你一个由 1&#xff08;陆地&#xff09;和 0&#xff08;水&#xff09;组成的的二维网格&#xff0c;请你计算网格中岛屿的数量。岛屿总是被水包围&#xff0c;并且每座岛屿只…...

阿里云专业翻译api对接

最近我们一个商城项目涉及多语言切换&#xff0c;默认中文。用户切换语言可选英语和阿拉伯语言&#xff0c;前端APP和后端返回动态数据都要根据用户选择语言来展示。前端静态内容都做了三套语言&#xff0c;后端商品为了适用这种多语言我们也进行了改造。每一件商品名称&#x…...

基于Spring Boot的能源管理系统+建筑能耗+建筑能耗监测系统+节能监测系统+能耗监测+建筑能耗监测

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

大数据新视界 --大数据大厂之 Cassandra 分布式数据库:高可用数据存储的新选择

&#x1f496;&#x1f496;&#x1f496;亲爱的朋友们&#xff0c;热烈欢迎你们来到 青云交的博客&#xff01;能与你们在此邂逅&#xff0c;我满心欢喜&#xff0c;深感无比荣幸。在这个瞬息万变的时代&#xff0c;我们每个人都在苦苦追寻一处能让心灵安然栖息的港湾。而 我的…...

ROS第五梯:ROS+VSCode+C++单步调试

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

SLA 概念和计算方法

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

C++比大小游戏

目录 开头程序程序的流程图程序游玩的效果下一篇博客要说的东西 开头 大家好&#xff0c;我叫这是我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 然后用单引号闭合&#xff1a; http://localhost/sqli-labs/Less-8/?id1 这关的问题在于报错是不显示&#xff0c;那没办法通过上篇文章的updatexml大法处理。对于这种情况&#xff0c;需要用“盲…...

Dijkstra算法和BFS算法(单源最短路径)

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

在WordPress中最佳Elementor主题推荐:专家级指南

对于已经在WordPress和Elementor上有丰富经验的用户来说&#xff0c;选择功能强大且高度灵活的主题&#xff0c;能大大提升网站的表现和定制能力。今天&#xff0c;我们来介绍六款适合用户的专家级Elementor主题&#xff1a;Sydney、Blocksy、Rife Free、Customify、Deep和Laye…...

关于RabbitMQ消息丢失的解决方案

RabbitMQ如何保证消息的可靠性传输 一、消息丢失的原因 1. 生产者端 网络问题&#xff1a; 原因&#xff1a;生产者与RabbitMQ服务器之间的网络连接不稳定或中断&#xff0c;导致消息在传输过程中丢失。解决方案&#xff1a;确保网络连接稳定&#xff0c;监控网络状态&#x…...

c语言动态内存分配

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

零基础制作一个ST-LINK V2 附PCB文件原理图 AD格式

资料下载地址&#xff1a;零基础制作一个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)配置简单&#xff0c;扩展性强(3)高可靠性(4)热部署(5)成本低、BSD许可证 1.4 Nginx的功能特性及常用功能基本HTTP服…...

监控系列之-Grafana面板展示及制作

一 Grafana设置添加数据源 1、设置Grafana中文显示 最后保存退出&#xff0c;数据源添加完毕 2、导入node_exporter主机监控面板 此处 有外网的情况下&#xff0c;直接输入对应面板的ID号&#xff0c;然后点击加载即可&#xff1b;无无外网的话&#xff0c;则考虑使用上传仪表…...

值传递和地址传递

值传递 我们从下面这段代码开始&#xff1a; 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 函数中&#xff0c;定义了一个字符数组 b 并…...

观成科技:隐蔽隧道工具Ligolo-ng加密流量分析

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

测试微信模版消息推送

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

React第五十七节 Router中RouterProvider使用详解及注意事项

前言 在 React Router v6.4 中&#xff0c;RouterProvider 是一个核心组件&#xff0c;用于提供基于数据路由&#xff08;data routers&#xff09;的新型路由方案。 它替代了传统的 <BrowserRouter>&#xff0c;支持更强大的数据加载和操作功能&#xff08;如 loader 和…...

2025年能源电力系统与流体力学国际会议 (EPSFD 2025)

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

中南大学无人机智能体的全面评估!BEDI:用于评估无人机上具身智能体的综合性基准测试

作者&#xff1a;Mingning Guo, Mengwei Wu, Jiarun He, Shaoxian Li, Haifeng Li, Chao Tao单位&#xff1a;中南大学地球科学与信息物理学院论文标题&#xff1a;BEDI: A Comprehensive Benchmark for Evaluating Embodied Agents on UAVs论文链接&#xff1a;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 用户&#xff0c;但你不希望用 root 权限运行 ns-3&#xff08;这是对的&#xff0c;ns3 工具会拒绝 root&#xff09;&#xff0c;你可以按以下方法创建一个 非 root 用户账号 并切换到它运行 ns-3。 一次性解决方案&#xff1a;创建非 roo…...

python如何将word的doc另存为docx

将 DOCX 文件另存为 DOCX 格式&#xff08;Python 实现&#xff09; 在 Python 中&#xff0c;你可以使用 python-docx 库来操作 Word 文档。不过需要注意的是&#xff0c;.doc 是旧的 Word 格式&#xff0c;而 .docx 是新的基于 XML 的格式。python-docx 只能处理 .docx 格式…...

HashMap中的put方法执行流程(流程图)

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