中间件——Kafka
两个系统各自都有各自要去做的事,所以只能将消息放到一个中间平台(中间件)
Kafka
分布式流媒体平台
程序发消息,程序接收消息
Producer:Producer即生产者,消息的产生者,是消息的入口。
Broker:Broker是kafka实例,每个服务器上有一个或多个kafka的实例,我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号,如图中的broker-0、broker-1等……
Topic:消息的主题,可以理解为消息的分类,kafka的数据就保存在topic。在每个broker上都可以创建多个topic。
Partition:Topic的分区,每个topic可以有多个分区,分区的作用是做负载,提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的,partition的表现形式就是一个一个的文件夹!
Replication:每一个分区都有多个副本,副本的作用是做备胎。当主分区(Leader)故障的时候会选择一个备胎(Follower)上位,成为Leader。在kafka中默认副本的最大数量是10个,且副本的数量不能大于Broker的数量,follower和leader绝对是在不同的机器,同一机器对同一个分区也只可能存放一个副本(包括自己)。
Message:每一条发送的消息主体。
Consumer:消费者,即消息的消费方,是消息的出口。
Consumer Group:我们可以将多个消费组组成一个消费者组,在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据,这也是为了提高kafka的吞吐量!
同一个组下,订阅的主题只能有一个消费者收到消息(一对一)
放到不同 的组下,就能实现一对多
Zookeeper:kafka集群依赖zookeeper来保存集群的的元信息,来保证系统的可用性。
Kafka对于zookeeper是清以来,保存kafka相关节点数据,管理节点。安装Kafka前需先安装zookeeper
1.Kafka生产者和消费者详解
Kafka设计与原理详解
生产者
消费者
谁增加while 让它一直处于监听状态
2.分区机制
加你个主题划分为多个分区(Partition)
可以处理更多的消息,不受单服务器限制,可以不受限的处理更多数据。
消息发送时都被发送到一个topic,其本质就是一个目录,而topic由是由一些Partition Logs(分区日志)组成,其组织结构如下图所示:
我们可以看到,每个Partition中的消息都是有序的,生产的消息被不断追加到Partition log上,其中的每一个消息都被赋予了一个唯一的offset值。
Kafka需要维持的元数据只有一个–消费消息在Partition中的offset值(偏移量-连续自增的数值),Consumer每消费一个消息,offset就会加1。其实消息的状态完全是由Consumer控制的,Consumer可以跟踪和重设这个offset值,这样的话Consumer就可以读取任意位置的消息。
把消息日志以Partition的形式存放有多重考虑,第一,方便在集群中扩展,每个Partition可以通过调整以适应它所在的机器,而一个topic又可以有多个Partition组成,因此整个集群就可以适应任意大小的数据了**;第二就是可以提高并发**,因为可以以Partition为单位读写了。
3.Kafka高可用机制
集群
备份
有了备份机制后,Kafka允许集群中的节点挂掉后而不影响整个集群工作。一个备份数量为n的集群允许n-1个节点失败。在所有备份节点中,有一个节点作为lead节点,这个节点保存了其它备份节点列表,并维持各个备份间的状体同步。
Kafka 提供了一种基于副本(Replication)的备份机制,以确保数据的高可用性和容错能力。以下是 Kafka 备份机制的详细说明:
-
1.副本(Replicas)与分区(Partitions)
在 Kafka 中,主题(Topic)被划分为多个分区(Partition),每个分区都有多个副本。Leader 副本负责处理所有对该分区的读写请求,而Follower 副本则从 Leader 副本同步数据。这样,即使某个 Broker(即 Leader 副本所在节点)发生故障,其他 Broker 上的 Follower 副本可以迅速晋升为新的 Leader,继续提供服务。 -
2.副本分配与复制
副本分配:Kafka 使用 ZooKeeper 管理元数据,包括分区与副本的分配信息。在创建主题时,可以指定每个分区的副本数(通常称为副本因子)。Kafka 会根据 Broker 配置和可用性,将分区的副本均匀地分布到不同的 Broker 上,以实现负载均衡和容错。
数据复制:Producer 发送消息到 Leader 副本。Leader 副本将消息写入其本地日志后,立即将消息发送给所有 Follower 副本。Follower 副本接收到消息后,将其写入本地日志。这种同步复制或异步复制(取决于配置)机制确保了数据在集群中的复制。
- 3.ISR(In-Sync Replicas)与副本同步
ISR:Kafka 维护了一个名为 ISR(In-Sync Replicas)的集合,包含所有与 Leader 副本保持同步的 Follower 副本。只有 I**SR 中的副本被认为是可以安全地晋升为 Leader 的候选者。**当 Follower 副本由于网络延迟、Broker 故障等原因与 Leader 副本失去同步时,会被暂时移出 ISR。
副本同步:Kafka 通过心跳机制监控 Follower 副本与 Leader 副本的同步状态。Follower 副本定期向 Leader 副本发送心跳,报告其已复制的消息偏移量。Leader 副本根据心跳信息判断 Follower 副本是否处于同步状态,并据此更新 ISR 集合。
**4. Leader 选举与故障恢复
Leader 选举:当 Leader 副本所在的 Broker 发生故障时,ZooKeeper 会检测到并触发 Leader 选举。从 ISR 集合中选择一个 Follower 副本晋升为新的 Leader。**其余 Follower 副本随后将与新的 Leader 建立连接并开始同步。
故障恢复:一旦新的 Leader 副本被选举出来,Producer 和 Consumer 可以无缝地切换到新的 Leader 进行读写操作。对于未完成同步的 Follower 副本,它们将在恢复连接后从新的 Leader 处拉取缺失的数据,直至重新加入 ISR。
4.消息可靠性
在消息系统中,保证消息在生产和消费过程中的可靠性是十分重要的,在实际消息传递过程中,可能会出现如下三中情况:
- 一个消息发送失败
- 一个消息被发送多次
- 最理想的情况:exactly-once ,一个消息发送成功且仅发送了一次
有许多系统声称它们实现了exactly-once,但是它们其实忽略了生产者或消费者在生产和消费过程中有可能失败的情况。比如虽然一个Producer成功发送一个消息,但是消息在发送途中丢失,或者成功发送到broker,也被consumer成功取走,但是这个consumer在处理取过来的消息时失败了。
(1)从Producer端看:Kafka是这么处理的,当一个消息被发送后,Producer会等待broker成功接收到消息的反馈(可通过参数控制等待时间),如果消息在途中丢失或是其中一个broker挂掉,Producer会重新发送(我们知道Kafka有备份机制,可以通过参数控制是否等待所有备份节点都收到消息)。
(2)从Consumer端看:前面讲到过partition,broker端记录了partition中的一个offset值,这个值指向Consumer下一个即将消费message。当Consumer收到了消息,但却在处理过程中挂掉,此时Consumer可以通过这个offset值重新找到上一个消息再进行处理。Consumer还有权限控制这个offset值,对持久化到broker端的消息做任意处理。
5.生产者详解
发送类型
异步时响应回调函数。
消息发送失败时会抛出异常,这是可以拿到异常,用于记录或做出补偿。
6.消费者详解
只有在一个分区,才能保证消息有序。
手动提示偏移量
同步提交尽可能会造成方法堵塞。
如果异步提交成功则会一直在while(true)监听中,如果出现异常才会捕获异常,捕获完异常后才会执行finally中的同步提交操作
7.Spring集成Kafka
异步,通过回调方式实现:
//发送消息
public void sendMessage(String message) {ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {@Overridepublic void onSuccess(SendResult<String, String> result) {System.out.println("Sent message=[" + message + "] with offset=[" + result.getRecordMetadata().offset() + "]");}@Overridepublic void onFailure(Throwable ex) {System.out.println("Unable to send message=[" + message + "] due to : " + ex.getMessage());}});
}
//接收消息
@KafkaListener(topics = "topicName", groupId = "foo")
public void listenGroupFoo(String message) {System.out.println("Received Message in group foo: " + message);
}
一个消费者可以监听多个主题:
@KafkaListener(topics = "topic1, topic2", groupId = "foo")
Spring也支持获取一个或多个消息头信息,通过在监听器上是哟个@Header注解:
@KafkaListener(topics = "topicName")
public void listenWithHeaders(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {System.out.println("Received Message: " + message" + "from partition: " + partition);
}
相关文章:

中间件——Kafka
两个系统各自都有各自要去做的事,所以只能将消息放到一个中间平台(中间件) Kafka 分布式流媒体平台 程序发消息,程序接收消息 Producer:Producer即生产者,消息的产生者,是消息的入口。 Brok…...

中介者模式(行为型)
目录 一、前言 二、中介者模式 三、总结 一、前言 中介者模式(Mediator Pattern)是一种行为型设计模式,又成为调停者模式,用一个中介对象来封装一系列的对象交互。中介者使各对象不需要显式地互相引用,从而使其耦合…...
定个小目标之刷LeetCode热题(45)
32. 最长有效括号 给你一个只包含 ( 和 ) 的字符串,找出最长有效(格式正确且连续)括号 子串的长度。 示例 1: 输入:s "(()" 输出:2 解释:最长有效括号子串是 "()"有事…...
golang 实现负载均衡器-负载均衡原理介绍
go 实现负载均衡器 文章目录 go 实现负载均衡器代码实现介绍负载均衡的核心组件与工作流程核心组件工作流程 总结 算法详细描述:1. 轮询(Round Robin)2. 最少连接(Least Connections)3. IP散列(IP Hash&…...

spring是如何解决循环依赖的,为什么不是两级
1. Spring使用三级缓存来解决循环依赖问题 Spring使用三级缓存来解决循环依赖问题,而不是使用两级缓存。 在Spring框架中,解决循环依赖的关键在于正确地管理Bean的生命周期和依赖关系。循环依赖指的是两个或多个Bean相互依赖,如果…...

大模型预训练优化参数设置
文章目录 基于批次数据的训练学习率优化器稳定优化技术与传统神经网络的优化类似,通常使用批次梯度下降算法来进行模型参数的调优。同时,通过调整学习率以及优化器中的梯度修正策略,可以进一步提升训练的稳定性。为了防止模型对数据产生过度拟合,训练中还需要引入一系列正则…...

PHP pwn 学习 (2)
文章目录 A. 逆向分析A.1 基本数据获取A.2 函数逆向zif_addHackerzif_removeHackerzif_displayHackerzif_editHacker A.3 PHP 内存分配 A.4 漏洞挖掘B. 漏洞利用B.1 PHP调试B.2 exp 上一篇blog中,我们学习了一些PHP extension for C的基本内容,下面结合一…...

【Python学习笔记】:Python爬取音频
【Python学习笔记】:Python爬取音频 背景前摇(省流可以不看): 人工智能公司实习,好奇技术老师训练语音模型的过程,遂请教,得知训练数据集来源于爬取某网页的音频。 很久以前看B站同济子豪兄的《…...

4 C 语言控制流与循环结构的深入解读
目录 1 复杂表达式的计算过程 2 if-else语句 2.1 基本结构及示例 2.2 if-else if 多分支 2.3 嵌套 if-else 2.4 悬空的 else 2.5 注意事项 2.5.1 if 后面不要加分号 2.5.2 省略 else 2.5.3 省略 {} 2.5.4 注意点 3 while 循环 3.1 一般形式 3.2 流程特点 3.3 注…...
vue排序
onEnd 函数示例,它假设 drag.value 是一个包含多个对象(每个对象至少包含 orderNum 和 label 属性)的数组,且您希望在拖动结束后更新所有元素的 orderNum 以反映新的顺序: function onEnd(e) { // 首先,确…...
agv叉车slam定位精度测试标准化流程
相对定位精度 条件:1.5m/s最高速度;基于普通直行任务 数据采集(3个不同位置的直行任务,每个任务直行约10m,每个10次) 测量每次走过的实际距离,与每次根据定位结果算得的相对距离,两…...

实战打靶集锦-31-monitoring
文章目录 1. 主机发现2. 端口扫描3. 服务枚举4. 服务探查4.1 ssh服务4.2 smtp服务4.3 http/https服务 5. 系统提权5.1 枚举系统信息5.2 枚举passwd文件5.3 枚举定时任务5.4 linpeas提权 6. 获取flag 靶机地址:https://download.vulnhub.com/monitoring/Monitoring.o…...

小程序-模板与配置
一、WXML模板语法 1.数据绑定 2.事件绑定 什么是事件 小程序中常用的事件 事件对象的属性列表 target和currentTarget的区别 bindtap的语法格式 在事件处理函数中为data中的数据赋值 3.事件传参与数据同步 事件传参 (以下为错误示例) 以上两者的…...
交叉编译aarch64的Qt5.12.2,附带Mysql插件编译
一、配置交叉编译工具链 1、交叉编译工具链目录 /opt/zlg/m3568-sdk-v1.0.0-ga/gcc-buildroot-9.3.0-2020.03-x86_64_aarch64-rockchip-linux-gnu/bin/aarch64-rockchip-linux-gnu-g /opt/zlg/m3568-sdk-v1.0.0-ga/gcc-buildroot-9.3.0-2020.03-x86_64_aarch64-rockchip-linu…...
好用的Ubuntu下的工具合集[持续增加]
1. 终端工具 UBUNTU下有哪些好用的终端软件? - 知乎 (zhihu.com) sudo apt install terminator...

Xcode 16 beta3 真机调试找不到 Apple Watch 的尝试解决
很多小伙伴们想用 Xcode 在 Apple Watch 真机上调试运行 App 时却发现:在 Xcode 设备管理器中压根找不到对应的 Apple Watch 设备。 大家是否已将 Apple Watch 和 Mac 都重启一万多遍了,还是束手无策。 Apple Watch not showing in XCodeApple Watch wo…...

Three.JS 使用RGBELoader和CubeTextureLoader 添加环境贴图
导入RGBELoader模块: import { RGBELoader } from "three/examples/jsm/loaders/RGBELoader.js"; 使用 addRGBEMappingk(environment, background,url) {rgbeLoader new RGBELoader();rgbeLoader.loadAsync(url).then((texture) > {//贴图模式 经纬…...
k8s logstash多管道配置
背景 采用的是标准的ELKfilebeat架构 ES版本:7.17.15 logstash版本:7.17.15 filebeat版本: 7.17.15 helm版本:7.17.3,官方地址:elastic/helm-charts 说一下为什么会想到使用多管道的原因 我们刚开始…...

【CMU博士论文】结构化推理增强大语言模型(Part 0)
问题 :语言生成和推理领域的快速发展得益于围绕大型语言模型的用户友好库的普及。这些解决方案通常依赖于Seq2Seq范式,将所有问题视为文本到文本的转换。尽管这种方法方便,但在实际部署中存在局限性:处理复杂问题时的脆弱性、缺乏…...
Odoo创建一个自定义UI视图
Odoo能够为给定的模型生成默认视图。在实践中,默认视图对于业务应用程序来说是绝对不可接受的。相反,我们至少应该以合乎逻辑的方式组织各个字段。 视图在带有Actions操作和Menus菜单的 XML 文件中定义。它们是模型的 ir.ui.view 实例。 列表视图 列表视…...

使用VSCode开发Django指南
使用VSCode开发Django指南 一、概述 Django 是一个高级 Python 框架,专为快速、安全和可扩展的 Web 开发而设计。Django 包含对 URL 路由、页面模板和数据处理的丰富支持。 本文将创建一个简单的 Django 应用,其中包含三个使用通用基本模板的页面。在此…...

3.3.1_1 检错编码(奇偶校验码)
从这节课开始,我们会探讨数据链路层的差错控制功能,差错控制功能的主要目标是要发现并且解决一个帧内部的位错误,我们需要使用特殊的编码技术去发现帧内部的位错误,当我们发现位错误之后,通常来说有两种解决方案。第一…...
渲染学进阶内容——模型
最近在写模组的时候发现渲染器里面离不开模型的定义,在渲染的第二篇文章中简单的讲解了一下关于模型部分的内容,其实不管是方块还是方块实体,都离不开模型的内容 🧱 一、CubeListBuilder 功能解析 CubeListBuilder 是 Minecraft Java 版模型系统的核心构建器,用于动态创…...

Vue2 第一节_Vue2上手_插值表达式{{}}_访问数据和修改数据_Vue开发者工具
文章目录 1.Vue2上手-如何创建一个Vue实例,进行初始化渲染2. 插值表达式{{}}3. 访问数据和修改数据4. vue响应式5. Vue开发者工具--方便调试 1.Vue2上手-如何创建一个Vue实例,进行初始化渲染 准备容器引包创建Vue实例 new Vue()指定配置项 ->渲染数据 准备一个容器,例如: …...

论文浅尝 | 基于判别指令微调生成式大语言模型的知识图谱补全方法(ISWC2024)
笔记整理:刘治强,浙江大学硕士生,研究方向为知识图谱表示学习,大语言模型 论文链接:http://arxiv.org/abs/2407.16127 发表会议:ISWC 2024 1. 动机 传统的知识图谱补全(KGC)模型通过…...
解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错
出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上,所以报错,到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本,cu、torch、cp 的版本一定要对…...
C++中string流知识详解和示例
一、概览与类体系 C 提供三种基于内存字符串的流,定义在 <sstream> 中: std::istringstream:输入流,从已有字符串中读取并解析。std::ostringstream:输出流,向内部缓冲区写入内容,最终取…...

前端开发面试题总结-JavaScript篇(一)
文章目录 JavaScript高频问答一、作用域与闭包1.什么是闭包(Closure)?闭包有什么应用场景和潜在问题?2.解释 JavaScript 的作用域链(Scope Chain) 二、原型与继承3.原型链是什么?如何实现继承&a…...

优选算法第十二讲:队列 + 宽搜 优先级队列
优选算法第十二讲:队列 宽搜 && 优先级队列 1.N叉树的层序遍历2.二叉树的锯齿型层序遍历3.二叉树最大宽度4.在每个树行中找最大值5.优先级队列 -- 最后一块石头的重量6.数据流中的第K大元素7.前K个高频单词8.数据流的中位数 1.N叉树的层序遍历 2.二叉树的锯…...

Unity | AmplifyShaderEditor插件基础(第七集:平面波动shader)
目录 一、👋🏻前言 二、😈sinx波动的基本原理 三、😈波动起来 1.sinx节点介绍 2.vertexPosition 3.集成Vector3 a.节点Append b.连起来 4.波动起来 a.波动的原理 b.时间节点 c.sinx的处理 四、🌊波动优化…...