kafka第三课-可视化工具、生产环境问题总结以及性能优化

news2025/2/2 23:18:47

一、可视化工具

https://pan.baidu.com/s/1qYifoa4 密码：el4o

  1. 下载解压之后，编辑该文件，修改zookeeper地址，也就是kafka注册的zookeeper的地址，如果是zookeeper集群，以逗号分开

vi conf/application.conf

在这里插入图片描述

  1. 启动命令
bin/kafka-manager
  1. 默认9000端口，直接访问，添加一个名字，名字只是方便开发者使用，随便起，地址是zookeeper的地址和端口
    在这里插入图片描述
    在这里插入图片描述
  2. 点进去可以看到对应的主体以及kafka的集群节点信息

在这里插入图片描述

二、线上环境规划

在这里插入图片描述
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。

这种大内存的情况一般都要用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偏移的消息开始消费，参见上节课的内容。

当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费
        latest(默认) ：只消费自己启动之后发送到主题的消息
        earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

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

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

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

在这里插入图片描述
平均值是1秒多，最大发送时间是1.9秒，不过这个时间都是一批消息的时间，百万数据kafka是分批发的，具体参考上一个文档，看他的缓存区和bash区
网络上很多资料都说分区数越多吞吐量越高 ， 但从压测结果来看，分区数到达某个值吞吐量反而开始下降，实际上很多事情都会有一个临界值，当超过这个临界值之后，很多原本符合既定逻辑的走向又会变得不同。一般情况分区数跟集群机器数量相当就差不多了。
当然吞吐量的数值和走势还会和磁盘、文件系统、 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的分布式事务需要额外开发功能。

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处理以及压缩传输

数据传输零拷贝原理：
在这里插入图片描述

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

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

相关文章

如何刻录光盘

如何刻录光盘

如何刻录光盘 1 、将光盘放入光驱&#xff0c;选择“用于CD/DVD播放机” &#xff0c;该模式下&#xff0c;刻录在光盘的文件无法进行编辑和删除 2 、将需要刻录的文件拷贝至光盘内&#xff0c;则会在“准备好写入光盘中的文件”下显示拷贝进去的文件&#xff0c;此时文件还没…
阅读更多...
EDI 工作流操作指南

EDI 工作流操作指南

一个完整的EDI工作流中&#xff0c;起始端为通常为文件传输端口&#xff1a;如AS2、OFTP等&#xff0c;末端为数据库端口。此前的文章中我们对AS2端口以及数据库端口已做了详细介绍&#xff0c;本文主要介绍 EDI 文件的格式转换以及映射。 如下图所示&#xff0c;工作流界面中…
阅读更多...
安装blissOS重启后无法进入图形化界面

安装blissOS重启后无法进入图形化界面

重启blissOS 重启时&#xff0c;按e键两下 进入 上图是一个可编辑页面&#xff0c;不要删除修改前面的内容&#xff0c;移动光标前往quiet&#xff0c;然后删除quiet输入“ nomodeset xforceseva ”&#xff0c;然后按下回车 然后按回车&#xff0c;按b键进入系统 在set-…
阅读更多...
class组件constructor方法

class组件constructor方法

class组件constructor方法 https://blog.csdn.net/m0_37557930/article/details/116228217 https://blog.csdn.net/qq_39207948/article/details/113143131 ​ 为何我们使用子类继承父类&#xff0c;就必须在 constructor( ) 方法中调用 super( ) 方法&#xff0c;否则新建实…
阅读更多...
【C】文件操作详解

【C】文件操作详解

这里写目录标题 文件操作什么是文件文件名文件类型文本文件二进制文件 文件缓冲区文件指针文件的打开和关闭fopenfclose 文件的顺序读写fgetcfputcfgetsfputcfscanffprintffwritefread比较scanf/fscanf/sscanfsscanf 比较printf/fprintf/sprintfsprintf 文件的随机读写fseekfte…
阅读更多...
周考一之重做

周考一之重做

输入一个学生的成绩&#xff0c;如果学习成绩>90分的同学用A表示&#xff0c;60-89分之间用B表示&#xff0c;60分以下的用C表示(10) public static void main(String[] args){ Scanner scanner new Scanner(System.in); System.out.println(“请输入学生成绩&#xff1a;…
阅读更多...
linux图形界面总结---X、Xorg、WM、QT、GTK、KDE、GNOME的区别与联系

linux图形界面总结---X、Xorg、WM、QT、GTK、KDE、GNOME的区别与联系

文章目录 一、 linux图形界面二、X协议三、Xfree86 Xorg四、WM(window manager:窗口管理器)五、X协议的Client端实现六、KDE、GNOME、QT和GTK直接关系七、参考&#xff1a; 一、 linux图形界面 linux本身没有图形界面&#xff0c;linux现在的图形界面的实现只是linux下的应用程…
阅读更多...
值类型与引用类型

值类型与引用类型

常见的值类型&#xff1a;int&#xff0c;long&#xff0c; short&#xff0c; float&#xff0c; double&#xff0c; byte&#xff0c; char&#xff0c; enum&#xff0c; struct...... 常见的引用类型&#xff1a;字符串&#xff0c; 数组&#xff0c; 类...... 区别&…
阅读更多...
附录10-3.JS正则常见案例

附录10-3.JS正则常见案例

目录 1 某一个字符串重复出现两次 2 多行字符串转变为数组套对象 3 多个正则表达式匹配一个字符 4 在指定的字符前加一些什么东西 1 某一个字符串重复出现两次 match的方式如果你使用小括号分组&#xff0c;第一个结果是符合正则的结果&#xff0c;也就是我查出来了 l…
阅读更多...
Excel表格怎么样转换成PDF？分享这3个方法免费转换！

Excel表格怎么样转换成PDF？分享这3个方法免费转换！

在日常办公和学习中&#xff0c;我们常常需要将Excel表格转换为PDF格式以便分享、打印或存档。本文将介绍三种简便的免费方法。方法一介绍了记灵在线工具&#xff0c;方法二使用办公软件&#xff08;WPS或Office&#xff09;&#xff0c;方法三则使用Adobe软件。 方法一&#…
阅读更多...
大二毕设.2-自研Spring框架

大二毕设.2-自研Spring框架

目录 项目描述&#xff1a; 基本演示 提取标记类 IOC容器的装载 IOC容器的操作 DI依赖注入 Aspect排序 AOP MVC 功能实现讲解 项目描述&#xff1a; 为了更好地学习 Spring 的核心&#xff0c;参考 Spring 源码实现的一个简易框架当前已实现 IOC&#xff0c;DI依赖注…
阅读更多...
交叉编译gRPC初实践

交叉编译gRPC初实践

快速开始 一、创建android编译目录&#xff0c;在grpc源码根目录下运行&#xff1a; mkdir -p cmake/build_android && cd cmake/build_android 二、cmkae生成对应Makefile等编译所需的文件 cmake -DCMAKE_TOOLCHAIN_FILE/zhuyazhou/DDS/tools_dds/android-ndk-r25/b…
阅读更多...
Python（Conda）环境迁移（从win10到macos12.5）笔记

Python（Conda）环境迁移（从win10到macos12.5）笔记

文章目录 背景环境 1、通过conda迁移2、通过python迁移3、最后&#xff08;逐一安装&#xff09; 背景环境 win10是以前安装的conda和py。目前需要导出的环境的版本为py3.10.4。macos是重新安装的conda&#xff0c;目前有的环境是py3.11.4。我是先进conda用刚安装好的base创建…
阅读更多...
idea 有时提示找不到类或者符号，日志报java: 找不到符号的解决

idea 有时提示找不到类或者符号，日志报java: 找不到符号的解决

解决一&#xff1a; idea maven编译成功&#xff0c;运行失败提示找不到符号&#xff0c;主要是get和set方法找不到符号&#xff0c;此时就是idea的lombok版本冲突 IDEA版本导致的Lombok失效&#xff0c;需要更新lombok版本到1.18.14及之后版本得到解决 <dependency>&…
阅读更多...
计划、逻辑与智能

计划、逻辑与智能

有计划性是指基于目标、目的或问题&#xff0c;通过制定计划、设立步骤和执行策略来达成预期结果的思维和行为。有计划的智能强调理性、逻辑和目标导向&#xff0c;它能够帮助人们更好地组织和管理资源&#xff0c;解决复杂的问题&#xff0c;并实现预期的目标。 无计划性则代表…
阅读更多...
Seal AppManager如何基于Terraform简化基础设施管理

Seal AppManager如何基于Terraform简化基础设施管理

作者简介 陈灿&#xff0c;数澈软件Seal 后端研发工程师&#xff0c;曾在腾讯负责敏捷研发体系建设以及 DevOps 解决方案的敏捷实践。在敏捷研发和产品效能提升有着丰富的经验&#xff0c;致力于构建一站式研发友好的平台工程解决方案。现在是 Seal 平台工程团队核心研发人员。…
阅读更多...
配置spark

配置spark

配置spark Yarn 模式Standalone 模式Local 模式 Yarn 模式 tar -zxvf spark-3.0.0-bin-hadoop3.2.tgz -C /opt/module cd /opt/module mv spark-3.0.0-bin-hadoop3.2 spark-yarn修改 hadoop 配置文件/opt/module/hadoop/etc/hadoop/yarn-site.xml, 并分发 <!--是否启动一…
阅读更多...
【板栗糖GIS】——如何安装ffmpeg

【板栗糖GIS】——如何安装ffmpeg

【板栗糖GIS】——如何安装ffmpeg 目录 1. 解压安装包 2. 把bin路径放在环境变量中 3. 检测是否安装成功 下载软件包&#xff0c;我已经准备好资源&#xff0c;只是审核还未通过&#xff0c;过两天会加上安装包链接 1. 解压安装包 2. 把bin路径放在环境变量中 3. 检测是否…
阅读更多...
Anaconda安装和激活

Anaconda安装和激活

一、Anaconda下载地址 https://mirrors.tuna.tsinghua.edu.cn/anaconda/archive/?CM&OD 说明&#xff1a;使用paddlepaddle需要先安装python环境&#xff0c;这里我们选择python集成环境Anaconda工具包 Anaconda是1个常用的python包管理程序安装完Anaconda后&#xff0c…
阅读更多...
面试经验分享 | 如何打造优秀的个人简历

面试经验分享 | 如何打造优秀的个人简历

月説小飞象交流会 一个人自身拥有越丰富&#xff0c;他对身外之物的需求也就越少&#xff0c;别人对他来说就越不重要。 内部交流│27期 面试经验分享| 打造优秀的个人简历 data analysis ●●●● 分享人&#xff1a;夏宇 暑假将至&#xff0c;毕业季来临&#xff0c;有准备实…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023