Kafka生产问题总结及性能优化实践

news2024/11/19 0:29:10

Kafka可视化管理工具kafka-manager

0

安装及基本使用可参考：https://www.cnblogs.com/dadonggg/p/8205302.html

线上环境规划

0

JVM参数设置

kafka是scala语言开发，运行在JVM上，需要对JVM参数合理设置，参看JVM调优专题

修改bin/kafka-start-server.sh中的jvm设置，假设机器是32G内存，可以如下设置：

export KAFKA_HEAP_OPTS="-Xmx16G -Xms16G -Xmn10G -XX:MetaspaceSize=256M -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:G1HeapRegionSize=16M"

这种大内存的情况一般都要用G1垃圾收集器，因为年轻代内存比较大，用G1可以设置GC最大停顿时间，不至于一次minor gc就花费太长时间，当然，因为像kafka，rocketmq，es这些中间件，写数据到磁盘会用到操作系统的page cache，所以JVM内存不宜分配过大，需要给操作系统的缓存留出几个G。

线上问题及优化

1、消息丢失情况：

消息发送端：

（1）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。大数据统计报表场景，对性能要求很高，对数据丢失不敏感的情况可以用这种。

（2）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。

（3）acks=-1或all： 这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。当然如果min.insync.replicas配置的是1则也可能丢消息，跟acks=1情况类似。

消息消费端：

如果消费这边配置的是自动提交，万一消费到数据还没处理完，就自动提交offset了，但是此时你consumer直接宕机了，未处理完的数据丢失了，下次也消费不到了。

2、消息重复消费

消息发送端：

发送消息如果配置了重试机制，比如网络抖动时间过长导致发送端发送超时，实际broker可能已经接收到消息，但发送方会重新发送消息

消息消费端：

如果消费这边配置的是自动提交，刚拉取了一批数据处理了一部分，但还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理

一般消费端都是要做消费幂等处理的。

3、消息乱序

如果发送端配置了重试机制，kafka不会等之前那条消息完全发送成功才去发送下一条消息，这样可能会出现，发送了1，2，3条消息，第一条超时了，后面两条发送成功，再重试发送第1条消息，这时消息在broker端的顺序就是2，3，1了

所以，是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息，当然acks不能设置为0，这样也能保证消息发送的有序。

kafka保证全链路消息顺序消费，需要从发送端开始，将所有有序消息发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低，可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。

4、消息积压

1）线上有时因为发送方发送消息速度过快，或者消费方处理消息过慢，可能会导致broker积压大量未消费消息。

此种情况如果积压了上百万未消费消息需要紧急处理，可以修改消费端程序，让其将收到的消息快速转发到其他topic(可以设置很多分区)，然后再启动多个消费者同时消费新主题的不同分区。

2）由于消息数据格式变动或消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未消费消息。

此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)，后面再慢慢分析死信队列里的消息处理问题。

5、延时队列

延时队列存储的对象是延时消息。所谓的“延时消息”是指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费，延时队列的使用场景有很多， 比如 ：

1）在订单系统中， 一个用户下单之后通常有 30 分钟的时间进行支付，如果 30 分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延时队列来处理这些订单了。

2）订单完成1小时后通知用户进行评价。

实现思路：发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中（topic_1s，topic_5s，topic_10s，...topic_2h，这个一般不能支持任意时间段的延时），然后通过定时器进行轮训消费这些topic，查看消息是否到期，如果到期就把这个消息发送到具体业务处理的topic中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了。

6、消息回溯

如果某段时间对已消费消息计算的结果觉得有问题，可能是由于程序bug导致的计算错误，当程序bug修复后，这时可能需要对之前已消费的消息重新消费，可以指定从多久之前的消息回溯消费，这种可以用consumer的offsetsForTimes、seek等方法指定从某个offset偏移的消息开始消费，参见上节课的内容。

7、分区数越多吞吐量越高吗

可以用kafka压测工具自己测试分区数不同，各种情况下的吞吐量

# 往test里发送一百万消息，每条设置1KB # throughput 用来进行限流控制，当设定的值小于 0 时不限流，当设定的值大于 0 时，当发送的吞吐量大于该值时就会被阻塞一段时间 bin/kafka-producer-perf-test.sh --topic test --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=192.168.65.60:9092 acks=1

0

网络上很多资料都说分区数越多吞吐量越高 ， 但从压测结果来看，分区数到达某个值吞吐量反而开始下降，实际上很多事情都会有一个临界值，当超过这个临界值之后，很多原本符合既定逻辑的走向又会变得不同。一般情况分区数跟集群机器数量相当就差不多了。

当然吞吐量的数值和走势还会和磁盘、文件系统、 I/O调度策略等因素相关。

注意：如果分区数设置过大，比如设置10000，可能会设置不成功，后台会报错"java.io.IOException : Too many open files"。

异常中最关键的信息是“ Too many open flies”，这是一种常见的 Linux 系统错误，通常意味着文件描述符不足，它一般发生在创建线程、创建 Socket、打开文件这些场景下 。 在 Linux系统的默认设置下，这个文件描述符的个数不是很多 ，通过 ulimit -n 命令可以查看：一般默认是1024，可以将该值增大，比如：ulimit -n 65535

8、消息传递保障

  • at most once(消费者最多收到一次消息，0--1次)：acks = 0 可以实现。
  • at least once(消费者至少收到一次消息，1--多次)：ack = all 可以实现。
  • exactly once(消费者刚好收到一次消息)：at least once 加上消费者幂等性可以实现，还可以用kafka生产者的幂等性来实现。

kafka生产者的幂等性：因为发送端重试导致的消息重复发送问题，kafka的幂等性可以保证重复发送的消息只接收一次，只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可，默认是false不开启。

具体实现原理是，kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker，broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查PID和Sequence Number，如果相同不会再接收。

PID：每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个PID 对用户完全是透明的。生产者如果重启则会生成新的PID。 Sequence Number：对于每个 PID，该 Producer 发送到每个 Partition 的数据都有对应的序列号，这些序列号是从0开始单调递增的。

9、kafka的事务

Kafka的事务不同于Rocketmq，Rocketmq是保障本地事务(比如数据库)与mq消息发送的事务一致性，Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)，一般在kafka的流式计算场景用得多一点，比如，kafka需要对一个topic里的消息做不同的流式计算处理，处理完分别发到不同的topic里，这些topic分别被不同的下游系统消费(比如hbase，redis，es等)，这种我们肯定希望系统发送到多个topic的数据保持事务一致性。Kafka要实现类似Rocketmq的分布式事务需要额外开发功能。

kafka的事务处理可以参考官方文档：

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("transactional.id", "my-transactional-id");
 Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
 //初始化事务
 producer.initTransactions();

 try {
     //开启事务
     producer.beginTransaction();
     for (int i = 0; i < 100; i++){
         //发到不同的主题的不同分区
         producer.send(new ProducerRecord<>("hdfs-topic", Integer.toString(i), Integer.toString(i)));
         producer.send(new ProducerRecord<>("es-topic", Integer.toString(i), Integer.toString(i)));
         producer.send(new ProducerRecord<>("redis-topic", Integer.toString(i), Integer.toString(i)));
     }
     //提交事务
     producer.commitTransaction();
 } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
     // We can't recover from these exceptions, so our only option is to close the producer and exit.
     producer.close();
 } catch (KafkaException e) {
     // For all other exceptions, just abort the transaction and try again.
     //回滚事务
     producer.abortTransaction();
 }
 producer.close();

10、kafka高性能的原因

  • 磁盘顺序读写：kafka消息不能修改以及不会从文件中间删除保证了磁盘顺序读，kafka的消息写入文件都是追加在文件末尾，不会写入文件中的某个位置(随机写)保证了磁盘顺序写。
  • 数据传输的零拷贝
  • 读写数据的批量batch处理以及压缩传输

数据传输零拷贝原理：

0

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

算法通关村第十二关黄金挑战——最长公共前缀问题解析

算法通关村第十二关黄金挑战——最长公共前缀问题解析

大家好&#xff0c;我是怒码少年小码。 最长公共前缀 LeetCode 14&#xff1a;编写一个函数来查找字符串数组中的最长公共前缀。 如果不存在公共前缀&#xff0c;返回空字符串 “”。 示例&#xff1a; 输入&#xff1a;strs [“flower”,“flow”,“flight”]输出&#xff…
阅读更多...
CSS+Javascript+Html日历控件

CSS+Javascript+Html日历控件

最近&#xff0c;因需要用HTMLJAVASCRIPTCSS实现了一个日历控件&#xff0c;效果如下&#xff1a; 单击上月、下月进行日历切换。当前日期在日历中变颜色标注显示。还是老老套路、老方法&#xff0c;分HMLCSSJAVASCRIPT三部分代码。 一、html代码 <h1>学习计划</h1…
阅读更多...
HubSpot CRM是什么？如何添加、使用呢？

HubSpot CRM是什么？如何添加、使用呢？

HubSpot CRM是一款强大的客户关系管理工具&#xff0c;它不仅简化了销售和市场营销过程&#xff0c;还提供了多种功能&#xff0c;有助于增强客户互动、提高销售效率和提供更多的洞察信息。 今天运营坛将带领大家深入了解HubSpot CRM&#xff0c;涵盖了它的定义、使用流程、添…
阅读更多...
国产芯片vs“国际水平”，有距离也有超越！

国产芯片vs“国际水平”，有距离也有超越！

当前&#xff0c;国产芯片正在迎来全新的发展阶段。国产终端芯片性能怎么样&#xff0c;与国际主流产品相比&#xff0c;表现如何&#xff1f;今天笔者就针对目前热度较高的四款国产CPU进行参数分析与性能跑分横向对比。 此次国产芯片评测型号分别是海光C86-3250、龙芯3A5000H…
阅读更多...
java蓝桥11-20题总结

java蓝桥11-20题总结

文章目录 11.排序1.代码2.知识点冒泡排序 12.跑步锻炼1.代码 13.蛇形填数1.代码 14.递增序列1.代码 15.AB&#xff08;OJ示例题目&#xff09;16.杨辉三角形1.代码2.知识点 17.货物1.代码 18.九进制转十进制1.代码 19.等差素数列1.代码 20.七段码 11.排序 1.代码 package lanq…
阅读更多...
网络套接字编程(一)

网络套接字编程(一)

网络套接字编程(一) 文章目录 网络套接字编程(一)预备知识源IP地址和目的IP地址端口号TCP/UDP协议特点网络字节序 socket编程socket常用APIsockaddr结构 简易UDP网络程序服务端创建套接字服务端绑定IP地址和端口号字符型IP地址VS整型IP地址服务端运行客户端创建套接字客户端绑定…
阅读更多...
python使用ffmpeg来制作音频格式转换工具（优化版）

python使用ffmpeg来制作音频格式转换工具（优化版）

简介：一个使用python加上ffmpeg模块来进行音频格式转换的工具。 日志： 20231030：第一版，设置了简单的UI布局和配色，实现音频转为Mp3、AAC、wav、flac四种格式。可解析音频并显示信息，可设置转换后的保存路径 UI界面： 编程平台：visual studio code 编程语言：python 3…
阅读更多...
如何把编辑器编辑好的文章分享给别人？

如何把编辑器编辑好的文章分享给别人？

哈喽&#xff0c;大家好&#xff0c;今天小编给大家分享一个非常实用的功能&#xff0c;当我们在使用微信编辑器编辑排版文章的时候&#xff0c;需要在不发表的前提下分享给好友或者需要先给领导看一下的时候该怎么办呢&#xff1f;有三个方法&#xff0c;都可以在不发表的前提…
阅读更多...
Ortec974A EPICS IOC程序

Ortec974A EPICS IOC程序

1&#xff09; 创建一个用户存放这个IOC程序结构的目录&#xff1a; rootorangepi4-lts:/usr/local/EPICS/program# mkdir ortec974A rootorangepi4-lts:/usr/local/EPICS/program# cd ortec974A/ rootorangepi4-lts:/usr/local/EPICS/program/ortec974A# ls2&#xff09;使用…
阅读更多...
系列四十二、Spring的事务传播行为案例演示（二）#REQUIRED

系列四十二、Spring的事务传播行为案例演示（二）#REQUIRED

一、演示Spring的默认传播行为&#xff08;REQUIRED&#xff09; 1.1、运行之前表中的数据 1.2、StockServiceImpl /*** Author : 一叶浮萍归大海* Date: 2023/10/30 15:43* Description:*/ Service(value "stockServiceREQUIRED") public class StockServiceImpl…
阅读更多...
Mysql设置了更新时间自动更新，指定更新部分sql时不进行时间更新

Mysql设置了更新时间自动更新，指定更新部分sql时不进行时间更新

现象&#xff1a; 因为字段设置了自动更新&#xff0c;所以sql语句一进行修改此字段就会自动更新时间&#xff0c;但是呢我们的有部分定时任务是半夜执行&#xff0c;并且不能让这个任务修改到数据的更新时间 解决&#xff1a; <update id"updateCreative">ALT…
阅读更多...
element-plus DateTimePicker日期选择器，限制指定日期和时间不可选择

element-plus DateTimePicker日期选择器，限制指定日期和时间不可选择

element-plus日期选择器&#xff0c;在指定日期时间前不可选择。 限制日期选择&#xff0c;使用disabled-date 限制小时选择&#xff0c;使用disabled-hours 限制分钟选择&#xff0c;使用disabled-minutes 限制毫秒选择&#xff0c;使用disabled-seconds 指定日期当天的时间有…
阅读更多...
一带一路10周年：爱创科技加速中国药企国际化征程

一带一路10周年：爱创科技加速中国药企国际化征程

“源自中国&#xff0c;属于世界”。 共建“一带一路”倡议提出10周年来&#xff0c;中国与沿线国家经济深度融合&#xff0c;在共商共建共享的基本原则下&#xff0c;“一带一路”形成了国际合作的平台和机制&#xff0c;跨国经济合作已基本形成。 随着“一带一路”合作日益加…
阅读更多...
【脚本笔记】AssetDatabase

【脚本笔记】AssetDatabase

AssetDatabase是编辑器下的处理资源操作的重要类&#xff0c;主要用于访问资源并针对资源执行操作的接口。 这里面所有的操作路径都是基于Unity项目的相对路径也就是Assets/xxx或者Assets/xxx.jpg这种。CacheServer 主要解决的是缩短大型团队导入资源的时间。当配置后&#xff…
阅读更多...
K8S部署应用时从harbor拉取镜像失败

K8S部署应用时从harbor拉取镜像失败

问题描述 K8S部署服务拉取镜像失败&#xff1a;ImagePullBackOff K8S拉取镜像提示&#xff1a;Failed to pull image “IP:PORT/zcy-project/nginx:1.16.1”: rpc error: code Unknown desc Error response from daemon: unauthorized: unauthorized to access repository: …
阅读更多...
mediapipe 训练自有图像数据分类

mediapipe 训练自有图像数据分类

参考&#xff1a; https://developers.google.com/mediapipe/solutions/customization/image_classifier https://colab.research.google.com/github/googlesamples/mediapipe/blob/main/examples/customization/image_classifier.ipynb#scrollToplvO-YmcQn5g 安装&#xff1a…
阅读更多...
配音软件怎么选？推荐三款自用好评的

配音软件怎么选？推荐三款自用好评的

会刷短视频的小伙伴肯定知道&#xff0c;很多创作者想让自己的视频更加丰富&#xff0c;呈现更加完美的视频效果&#xff0c;往往会在视频里的空境部分加入一些旁白解说&#xff0c;你们难道就不好奇吗&#xff1f;这些声音为什么这么想真人说的话&#xff0c;而且还可以这么自…
阅读更多...
虚幻C++基础 day1

虚幻C++基础 day1

虚幻C概念 虚幻C类的继承结构 虚幻引擎C类层级结构(Hierarchy) 这些基本类又派生出了很多子类&#xff0c;例&#xff1a; UE中的反射与垃圾回收系统 例如一个创建了一个Actor类&#xff0c;有一个Actor类型指针去指向这个Actor类&#xff0c;如果的指针被销毁了&#xff…
阅读更多...
编译环境里存在yaml-cpp的多个版本时可能引起的问题

编译环境里存在yaml-cpp的多个版本时可能引起的问题

有时要编译的程序自带了特定版本的yaml-cpp&#xff0c;同时系统目录下也安装了更高版本的yaml-cpp&#xff0c;这时可能引起编译错误&#xff0c;就是某些yaml-cpp的API不认识&#xff0c;例如&#xff1a; 出现这种问题倒好办&#xff0c;正常情况下不可能&#xff0c;肯定能…
阅读更多...
目标检测算法发展史

目标检测算法发展史

前言 比起图像识别&#xff0c;现在图片生成技术要更加具有吸引力&#xff0c;但是要步入AIGC技术领域&#xff0c;首先不推荐一上来就接触那些已经成熟闭源的包装好了再提供给你的接口网站&#xff0c;会使用别人的模型生成一些图片就能叫自己会AIGC了吗&#xff1f;那样真正…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023