消息中间件篇之Kafka-消息不丢失

一、 正常工作流程

        生产者发送消息到kafka集群,然后由集群发送到消费者。

        但是可能中途会出现消息的丢失。下面是解决方案。

二、 生产者发送消息到Brocker丢失

1. 设置异步发送

    //同步发送RecordMetadata recordMetadata = kafkaProducer.send(record).get();//异步发送kafkaProducer.send(record,new Callback() {@Override public void onCompletion (RecordMetadata recordMetadata, Exception e){if (e != null) {System.out.println("消息发送失败 | 记录日志");}long offset = recordMetadata.offset();int partition = recordMetadata.partition();String topic = recordMetadata.topic();}});

2.消息重试

//设置重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);

三、消息在Brocker中存储丢失

        发送确认机制acks。消息首先Topic是key,到达Topic以后才选择分区Partition(默认就一个分区,0号分区),默认连接的就是分区的Leader节点,由leader分区同步到follower区中。

四、消费者从Brocker接收消息丢失

1.分区机制

        1. Kafka 中的分区机制指的是将每个主题划分成多个分区(Partition)。

        2. topic分区中消息只能由消费者组中的唯一一个消费者处理,不同的分区分配给不同的消费者(同一个消费者组)。

2.消费方式

        消费者默认是自动按期提交已经消费的偏移量,默认是每隔5s提交一次,如果出现重平衡的情况,可能会重复消费或丢失数据

3.那如何解决重复消费

        禁用自动提交偏移量,改为手动: 1. 同步提交。  2. 异步提交。 3. 同步+异步组合提交。

       

五、面试题

面试官:Kafka是如何保证消息不丢失?

候选人:嗯,这个保证机制很多,在发送消息到消费者接收消息,在每个阶段都有可能会丢失消息,所以我们解决的话也是从多个方面考虑:

第一个是生产者发送消息的时候,可以使用异步回调发送,如果消息发送失败,我们可以通过回调获取失败后的消息信息,可以考虑重试或记录日志,后边再做补偿都是可以的。同时在生产者这边还可以设置消息重试,有的时候是由于网络抖动的原因导致发送不成功,就可以使用重试机制来解决。

第二个在broker中消息有可能会丢失,我们可以通过kafka的复制机制来确保消息不丢失,在生产者发送消息的时候,可以设置一个acks,就是确认机制。我们可以设置参数为all,这样的话,当生产者发送消息到了分区之后,不仅仅只在leader分区保存确认,在follwer分区也会保存确认,只有当所有的副本都保存确认以后才算是成功发送了消息,所以,这样设置就很大程度了保证了消息不会在broker丢失。

第三个有可能是在消费者端丢失消息,kafka消费消息都是按照offset进行标记消费的,消费者默认是自动按期提交已经消费的偏移量,默认是每隔5s提交一次,如果出现重平衡的情况,可能会重复消费或丢失数据。我们一般都会禁用掉自动提价偏移量,改为手动提交,当消费成功以后再报告给broker消费的位置,这样就可以避免消息丢失和重复消费了。

面试官:Kafka中消息的重复消费问题如何解决的?

候选人:kafka消费消息都是按照offset进行标记消费的,消费者默认是自动按期提交已经消费的偏移量,默认是每隔5s提交一次,如果出现重平衡的情况,可能会重复消费或丢失数据。我们一般都会禁用掉自动提价偏移量,改为手动提交,当消费成功以后再报告给broker消费的位置,这样就可以避免消息丢失和重复消费了。

为了消息的幂等,我们也可以设置唯一主键来进行区分,或者是加锁,数据库的锁,或者是redis分布式锁,都能解决幂等的问题。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/265534.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

机器学习:SVM算法(Python)

一、核函数 kernel_func.py import numpy as npdef linear():"""线性核函数:return:"""def _linear(x_i, x_j):return np.dot(x_i, x_j)return _lineardef poly(degree3, coef01.0):"""多项式核函数:param degree: 阶次:param …

K8S-001-Virtual box - Network Config

A. 配置两个IP, 一个连接内网,一个链接外网: 1. 内网配置(Host only, 不同的 virutal box 的版本可以不一样,这些窗口可能在不同的地方,但是配置的内容是一样的): 静态IP 动态IP 2. 外网(创建一个 Networ…

【Linux】进程优先级以及Linux内核进程调度队列的简要介绍

进程优先级 基本概念查看系统进程修改进程的优先级Linux2.6内核进程调度队列的简要介绍和进程优先级有关的概念进程切换 基本概念 为什么会存在进程优先级?   进程优先级用于确定在资源竞争的情况下,哪个进程将被操作系统调度为下一个运行的进程。进程…

【Ubuntu】使用WSL安装Ubuntu

WSL 适用于 Linux 的 Windows 子系统 (WSL) 是 Windows 的一项功能,可用于在 Windows 计算机上运行 Linux 环境,而无需单独的虚拟机或双引导。 WSL 旨在为希望同时使用 Windows 和 Linux 的开发人员提供无缝高效的体验。安装 Linux 发行版时&#xff0c…

数据库架构师之道:MySQL安装与系统整合指南

目录 MySQL数据库安装(centos) 版本选择 企业版 社区版 选哪个 MySQL特点 MySQL服务端-客户端 mysql下载选择 软件包解释 安装MySQL的方式 rpm包安装 yum方式安装 源码编译安装★ 具体的编译安装步骤★★ 环境准备 free -m命令 cat /pr…

基于java Springboot实现课程评分系统设计和实现

基于java Springboot实现课程评分系统设计和实现 博主介绍:多年java开发经验,专注Java开发、定制、远程、文档编写指导等,csdn特邀作者、专注于Java技术领域 作者主页 央顺技术团队 Java毕设项目精品实战案例《1000套》 欢迎点赞 收藏 ⭐留言 文末获取源…

【寸铁的刷题笔记】图论、bfs、dfs

【寸铁的刷题笔记】图论、bfs、dfs 大家好 我是寸铁👊 金三银四,图论基础结合bfs、dfs是必考的知识点✨ 快跟着寸铁刷起来!面试顺利上岸👋 喜欢的小伙伴可以点点关注 💝 🌞详见如下专栏🌞 &…

我在代码随想录|写代码Day27 | 贪心算法 | 122.买卖股票的最佳时机 II,55. 跳跃游戏, 45.跳跃游戏 II

&#x1f525;博客介绍&#xff1a; 27dCnc &#x1f3a5;系列专栏&#xff1a; <<数据结构与算法>> << 算法入门>> << C项目>> &#x1f3a5; 当前专栏: <<数据结构与算法>> 专题 : 数据结构帮助小白快速入门算法 &…

代码随想录算法训练营29期|day64 任务以及具体安排

第十章 单调栈part03 有了之前单调栈的铺垫&#xff0c;这道题目就不难了。 84.柱状图中最大的矩形class Solution {int largestRectangleArea(int[] heights) {Stack<Integer> st new Stack<Integer>();// 数组扩容&#xff0c;在头和尾各加入一个元素int [] ne…

算法沉淀——动态规划之路径问题(leetcode真题剖析)

算法沉淀——动态规划之路径问题 01.不同路径02.不同路径 II03.珠宝的最高价值04.下降路径最小和05.最小路径和06.地下城游戏 01.不同路径 题目链接&#xff1a;https://leetcode.cn/problems/unique-paths/ 一个机器人位于一个 m x n 网格的左上角 &#xff08;起始点在下图…

鸿运(通天星CMSV6车载)主动安全监控云平台敏感信息泄露漏洞

文章目录 前言声明一、系统简介二、漏洞描述三、影响版本四、漏洞复现五、修复建议 前言 鸿运主动安全监控云平台实现对计算资源、存储资源、网络资源、云应用服务进行7*24小时全时区、多地域、全方位、立体式、智能化的IT运维监控&#xff0c;保障IT系统安全、稳定、可靠运行…

Mycat核心教程--Mycat 监控工具【四】

Mycat核心教程--Mycat 监控工具 九、Mycat 监控工具9.1.Mycat-web 简介9.2.Mycat-web 配置使用9.2.1.ZooKeeper 安装【上面有】9.2.2.Mycat-web 安装9.2.2.1.下载安装包9.2.2.2.安装包拷贝到Linux系统/opt目录下&#xff0c;并解压9.2.2.3.拷贝mycat-web文件夹到/usr/local目录…

堆和堆排序【数据结构】

目录 一、堆1. 堆的存储定义2. 初始化堆3. 销毁堆4. 堆的插入向上调整算法 5. 堆的删除向下调整算法 6. 获取堆顶数据7. 获取堆的数据个数8. 堆的判空 二、Gif演示三、 堆排序1. 堆排序(1) 建大堆(2) 排序 2.Topk问题 四、完整代码1.堆的代码Heap.cHeap.htest.c 2. 堆排序的代码…

最新IE跳转Edge浏览器解决办法(2024.2.26)

最新IE跳转Edge浏览器解决办法&#xff08;2024.2.26&#xff09; 1. IE跳转原因1.1. 原先解决办法1.2. 最新解决办法1.3. 最后 1. IE跳转原因 关于IE跳转问题是由于在2023年2月14日&#xff0c;微软正式告别IE浏览器&#xff0c;导致很多使用Windows10系统的电脑在打开IE浏览…

树莓派 关闭低电压闪电报警和文字报警

关闭低电压闪电图标报警 方法&#xff1a; sudo nano /boot/config.txt在末尾加上 avoid_warnings1重启就可以了 关闭文字报警 方法&#xff1a; sudo apt remove lxplug-ptbatt然后重启就可以了

【论文阅读】基于人工智能目标检测与跟踪技术的过冷流沸腾气泡特征提取

Bubble feature extraction in subcooled flow boiling using AI-based object detection and tracking techniques 基于人工智能目标检测与跟踪技术的过冷流沸腾气泡特征提取 期刊信息&#xff1a;International Journal of Heat and Mass Transfer 2024 级别&#xff1a;EI检…

SpringCloud-Gateway解决跨域问题

Spring Cloud Gateway是一个基于Spring Framework的微服务网关&#xff0c;用于构建可扩展的分布式系统。在处理跨域问题时&#xff0c;可以通过配置网关来实现跨域资源共享&#xff08;CORS&#xff09;。要解决跨域问题&#xff0c;首先需要在网关的配置文件中添加相关的跨域…

SNMP简介

定义 简单网络管理协议SNMP&#xff08;Simple Network Management Protocol&#xff09;是广泛应用于TCP/IP网络的网络管理标准协议。SNMP提供了一种通过运行网络管理软件的中心计算机&#xff08;即网络管理工作站&#xff09;来管理设备的方法。SNMP的特点如下&#xff1a;…

Python爬虫获取淘宝商品详情页数据|实现自动化采集商品信息

要实现自动化采集淘宝商品详情页数据&#xff0c;可以使用Python的第三方库如requests和BeautifulSoup。以下是一个简单的示例&#xff1a; Taobao.item_get-获得淘宝商品详情数据接口返回值说明 1.请求方式:HTTP POST &#xff1b;复制Taobaoapi2014获取APISDK文件。 2.请求…

如何让网页APP化 渐进式Web应用(PWA)

前言 大家上网应该发现有的网页说可以安装对应应用&#xff0c;结果这个应用好像就是个web&#xff0c;不像是应用&#xff0c;因为这里采用了PWA相关技术。 PWA&#xff0c;全称为渐进式Web应用&#xff08;Progressive Web Apps&#xff09;&#xff0c;是一种可以提供类似…