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

一、可视化工具

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.rhkb.cn/news/73211.html

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

相关文章

Rust 数据类型 之 结构体(Struct)

目录 结构体&#xff08;Struct&#xff09; 定义与声明 结构体定义 结构体实例 结构体分类 单元结构体&#xff08;Unit Struct&#xff09; 元组结构体&#xff08;Tuple Struct&#xff09; 具名结构体&#xff08;Named Struct&#xff09; 结构体嵌套 结构体方法…

公网访问的Linux CentOS本地Web站点搭建指南

文章目录 前言1. 本地搭建web站点2. 测试局域网访问3. 公开本地web网站3.1 安装cpolar内网穿透3.2 创建http隧道&#xff0c;指向本地80端口3.3 配置后台服务 4. 配置固定二级子域名5. 测试使用固定二级子域名访问本地web站点 前言 在web项目中,部署的web站点需要被外部访问,则…

总结946

6:40起床 7&#xff1a;15~8:00早读&#xff0c;07年tex1,2 8:10~10:12 880第二章选填&#xff0c;题目有些综合&#xff0c;错的有些多呀&#xff0c;不要紧&#xff0c;拿下它&#xff0c;就有进步了。 10:28~11:27重做强化18讲6道题 12&#xff1a;10~2:15吃饭睡觉&…

Python实现GA遗传算法优化循环神经网络分类模型(LSTM分类算法)项目实战

说明&#xff1a;这是一个机器学习实战项目&#xff08;附带数据代码文档视频讲解&#xff09;&#xff0c;如需数据代码文档视频讲解可以直接到文章最后获取。 1.项目背景 遗传算法&#xff08;Genetic Algorithm&#xff0c;GA&#xff09;最早是由美国的 John holland于20世…

chatgpt赋能python:如何让Python暂停?

如何让Python暂停&#xff1f; Python是一种高级编程语言&#xff0c;常用于数据分析、机器学习等领域。在Python编程中&#xff0c;我们经常需要让程序执行暂停一段时间&#xff0c;等待某些操作完成。本文将介绍如何让Python暂停&#xff0c;以及如何在SEO中优化文章标题&am…

分享 7 个不错的 AI 工具

人工智能的世界继续让我们着迷&#xff0c;近期的 OpenAI ChatGPT 掀起人们对人工智能的更大的期待&#xff0c;本文收集了 7 个人工智能 (AI) 工具&#xff0c;其中大部分易于使用&#xff0c;有些更复杂……比如构建 ML 模型。 1. GFP-GAN&#xff1a;照片修复 GFP-GAN 是一…

世界杯决赛解析

新体育 2023-01-04 10:03 发表于北京 卡塔尔世界杯决赛跌宕起伏&#xff0c;精彩纷呈。双方主帅斗智斗勇&#xff0c;妙手迭出&#xff0c;奉献了一场难得一见的对攻大战。赛后回顾&#xff0c;阿根廷的斯卡洛尼和法国的德尚用兵有很多值得学习领悟之处。从战术的角度看&#x…

谈一谈我心中的世界杯

2022卡塔尔世界杯 开赛在即 不论你喜不喜欢足球 恐怕都无法脱离 世界杯带来的影响 如果不能和人随时随地 聊上几句世界杯话题 那得多尴尬 有了这份“伪球迷速成指南” 一定能帮助你 在各种尬聊场合 脱颖而出↓↓ 1. 世界杯的由来 世界杯每4年举办一次。世界杯又称生…

十分钟带你玩转人工智能——调用百度AI接口实现文字转语音

调用别人的接口&#xff0c;实现人工智能就是站在巨人的肩膀上 打开百度AI&#xff0c;点这个控制台&#xff0c;&#xff08;你要是没有注册 &#xff0c;就注册一下&#xff0c;很简单的&#xff09; 点开这个语音技术 创建一下应用 好了以后&#xff0c;按照这个图的步…

含辞未吐,声若幽兰,史上最强免费人工智能AI语音合成TTS服务微软Azure(Python3.10接入)

所谓文无第一&#xff0c;武无第二&#xff0c;云原生人工智能技术目前呈现三足鼎立的态势&#xff0c;微软&#xff0c;谷歌以及亚马逊三大巨头各擅胜场&#xff0c;不分伯仲&#xff0c;但目前微软Azure平台不仅仅只是一个PaaS平台&#xff0c;相比AWS&#xff0c;以及GAE&am…

对接百度文心一言API---人工智能工作笔记0039

1.获取文心一言的api key 这个地址然后 点击 创建API key就可以了 然后先用postman测试一下 https://wenxin.baidu.com/moduleApi/portal/api/oauth/tokenclient_credentials application/x-www-form-urlencoded 注意 这里的参数grant_type这个是client_credentials是固定的…

斩获23项冠军,日均调用破万亿!百度交出年度AI成绩单:语音语言领衔技术突破,国产自研成大趋势...

乾明 雷刚 发自 凹非寺 量子位 报道 | 公众号 QbitAI 刚刚&#xff0c;百度一年AI成绩单官宣。 作为中国AI头雁&#xff0c;百度的成绩单不止代表巨头公司业绩&#xff0c;也代表了技术前沿现状和产业基本情况。 而且诸多进展&#xff0c;也是全新起跑线上全球AI发展的侧影&am…

AI百科:一个开启人工智能时代的综合性平台

无论是人工智能的快速发展还是AI技术在各个领域的广泛应用&#xff0c;都让我们对智能未来充满了期待和好奇。随着科技的进步&#xff0c;发现了一个好网站&#xff0c;一个集合了丰富AI工具和产品介绍的综合性网站—— AI百科。 在人工智能&#xff08;AI&#xff09;领域的快…

三大运营商乘风破浪,为什么离不开BAT等互联网企业?

【全球云观察 &#xff5c; 热点关注】对于中国联通从2017年开始的混改&#xff0c;业界关注挺多&#xff0c;之后确实带入了互联网企业如腾讯、京东、阿里、百度等参与&#xff0c;从而对中国联通在云计算、物联网等新兴技术领域的发展激发了业务的创新热情。 特别是2022年11…

周鸿祎:腾讯将来可能会是中国最大的运营商

近日&#xff0c;本刊记者采访了周鸿祎近一个小时&#xff0c;探讨未来互联网的变化、颠覆及未来&#xff0c;周鸿祎怎么看待互联网巨头的崛起对传统产业的影响和变化?在巨头林立的时代怎样才能颠覆式创新打败对手?如何选择你的对手和伙伴&#xff0c;周鸿祎说&#xff0c;“…

进程地址空间详解

进程地址空间 文章目录 进程地址空间验证地址空间的基本排布进程地址空间究竟是什么&#xff1f; 地址空间和物理内存之间的关系为什么要存在地址空间&#xff1f; 我们在学习C语言期间&#xff0c;经常可以提及到这些区域&#xff0c;有一个问题&#xff1a;这里的地址空间是内…

【GlobalMapper精品教程】044:空间操作(1)——交集(Intersection)

GlobalMapper提供的空间分析(操作)的方法有:交集、并集、单并集、差异、对称差集、相交、重叠、接触、包含、等于、内部、分离等,本文主要讲述交集工具的使用。 文章目录 一、实验数据二、符号化设置三、交集运算四、结果展示1. 运行提示2. 空间查看3. 属性表查看五、心灵感…

PostgreSQL的表空间

PostgreSQL之表空间 1、什么是PG中的表空间&#xff08;tablespace&#xff09;&#xff1f; pg中的表空间实际上就是文件系统中的一个目录, 是pg中数据库对象&#xff08;包括表、索引等&#xff09;的容器。 Pg使用操作系统的文件系统进行对象存储&#xff0c;每个数据库对…

Oracle表空间、用户详解

目录 新建连接三者关系表空间创建表空间修改表空间和数据文件修改数据文件容量新增表空间的数据文件重命名数据文件修改表空间状态修改数据文件状态 删除表空间查询 用户创建删除查询修改 新建连接 工具选择&#xff1a; 我们一般会选择一个工具来连接本地的Oracle&#xff0c…

49-Linux_malloc及4G虚拟空间分布

文章目录 malloc及4G虚拟空间分布1.malloc2.32位操作系统进程的4G虚拟空间分布1) 代码区(Text egment)&#xff1a;2) 全局初始化数据区/静态数据区(Data Segment)3) 未初始化数据区(BSS)&#xff1a;4) 堆区(heap)&#xff1a;5) 栈区(stack)&#xff1a; malloc及4G虚拟空间分…