做猎头顾问 经常看哪些网站,网站运营的含义,专题探索网站开发模式特点,长沙推广公司Kafka与时间轮
Kafka中存在大量的延时操作。 1、发送消息-超时重试机制
2、ACKS 用于指定分区中必须要有多少副本收到这条消息#xff0c;生产者才认为写入成功#xff08;延时 等#xff09;
Kafka并没有使用JDK自带的Timer或者DelayQueue来实现延迟的功能#xff0c;而…Kafka与时间轮
Kafka中存在大量的延时操作。 1、发送消息-超时重试机制
2、ACKS 用于指定分区中必须要有多少副本收到这条消息生产者才认为写入成功延时 等
Kafka并没有使用JDK自带的Timer或者DelayQueue来实现延迟的功能而是基于时间轮自定义了一个用于实现延迟功能的定时器SystemTimer
JDK的Timer和DelayQueue插入和删除操作的平均时间复杂度为O(log(n))并不能满足Kafka的高性能要求而基于时间轮可以将插入和删除操作的时间复杂度都降为O(1)。
时间轮的应用并非Kafka独有其应用场景还有很多在Netty、Akka、Quartz、Zookeeper等组件中都存在时间轮的踪影。
时间轮
Java中任务调度
要回答这个问题我们先从Java中最原始的任务调度的方法说起。
给你一批任务假设有1000个任务都是不同的时间执行的时间精确到秒你怎么实现对所有的任务的调度
第一种思路是启动一个线程每秒钟对所有的任务进行遍历找出执行时间跟当前时间匹配的执行它。如果任务数量太大遍历和比较所有任务会比较浪费时间。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cu92Jn2L-1690353345189)(file:///C:/Users/root/AppData/Local/Temp/ksohtml10964/wps14.jpg)]
第二个思路把这些任务进行排序执行时间近先触发的放在前面。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GfOAIem8-1690353345192)(file:///C:/Users/root/AppData/Local/Temp/ksohtml10964/wps15.jpg)]
如果是数组的时间的话这里会涉及到大量的元素移动新加入任务任务执行–删除任务之类都需要重新排序
那么在Java代码怎么实现呢
JDK包里面自带了一个Timer工具类java.util包下可以实现延时任务例如30分钟以后触发也可以实现周期性任务例如每1小时触发一次。
它的本质是一个优先队列TaskQueue和一个执行任务的线程TimerThread。
普通的队列是一种先进先出的数据结构元素在队列尾追加而从队列头删除。在优先队列中元素被赋予优先级。当访问元素时具有最高优先级的元素最先删除。优先队列具有最高级先出 first in, largest out的行为特征。通常采用堆数据结构来实现。 在这个优先队列中最先需要执行的任务排在优先队列的第一个。然后 TimerThread 不断地拿第一个任务的执行时间和当前时间做对比。如果时间到了先看看这个任务是不是周期性执行的任务如果是则修改当前任务时间为下次执行的时间如果不是周期性任务则将任务从优先队列中移除。最后执行任务。
但是Timer是单线程的在很多场景下不能满足业务需求。
在JDK1.5之后引入了一个支持多线程的任务调度工具ScheduledThreadPoolExecutor用来替代TImer它是几种常用的线程池之一。看看构造函数里面是一个延迟队列DelayedWorkQueue也是一个优先队列。 DelayedWorkQueue的最小堆实现
优先队列的使用的是最小堆实现。
最小堆的含义: 一种完全二叉树, 父结点的值小于或等于它的左子节点和右子节点
比如插入以下的数据 [1,2,3,7,17,19,25,36,100]
最小堆就长成这个样子。 优先队列的插入和删除的时间复杂度是O(logn)当数据量大的时候频繁的入堆出堆性能不是很好。
比如要插入0过程如下
1、插入末尾元素 2、0比19小所以要向上移动且互换。 3、0比2小所以要向上移动且互换。 4、0比2小所以要向上移动且互换。 算法复杂度
N个数据的最小堆, 共有logN层, 最坏的情况下, 需要移动logN次
时间轮
这里我们先考虑对所有的任务进行分组把相同执行时刻的任务放在一起。比如这里数组里面的一个下标就代表1秒钟。它就会变成一个数组加链表的数据结构。分组以后遍历和比较的时间会减少一些。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dpk4Jicj-1690353345195)(file:///C:/Users/root/AppData/Local/Temp/ksohtml10964/wps16.jpg)]
但是还是有问题如果任务数量非常大而且时间都不一样或者有执行时间非常遥远的任务那这个数组长度是不是要非常地长比如有个任务2个月之后执行从现在开始计算它的下标是5253120。
所以长度肯定不能是无限的只能是固定长度的。比如固定长度是8一个格子代表1秒现在叫做一个bucket槽一圈可以表示8秒。遍历的线程只要一个格子一个格子的获取任务并且执行就OK了。
固定长度的数组怎么用来表示超出最大长度的时间呢可以用循环数组。
比如一个循环数组长度8可以表示8秒。8秒以后执行的任务怎么放进去只要除以8用得到的余数放到对应的格子就OK了。比如10%82它放在第2个格子。这里就有了轮次的概念第10秒的任务是第二轮的时候才执行。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qCMzQcjf-1690353345196)(file:///C:/Users/root/AppData/Local/Temp/ksohtml10964/wps17.jpg)]
这时候时间轮的概念已经出来了。
如果任务数量太多相同时刻执行的任务很多会导致链表变得非常长。这里我们可以进一步对这个时间轮做一个改造做一个多层的时间轮。
比如最内层8个格子每个格子1秒外层8个格子每个格子8*864秒最内层走一圈外层走一格。这时候时间轮就跟时钟更像了。随着时间流动任务会降级外层的任务会慢慢地向内层移动。 时间轮任务插入和删除时间复杂度都为O(1)应用范围非常广泛更适合任务数很大的延时场景。Dubbo、Netty、Kafka中都有实现。
Kafka中时间轮实现
Kafka里面TimingWheel的数据结构 kafka会启动一个线程去推动时间轮的指针转动。其实现原理其实就是通过queue.poll()取出放在最前面的槽的TimerTaskList 添加新的延迟任务 往时间轮添加新的任务 时间轮指针的推进 第二层时间轮的创建代码如下 Kafka性能问题
1、kafka如何确保消息的可靠性传输
这个问题需要从以下3个方面分析和解决
1消费端弄丢了数据
唯一可能导致消费者弄丢数据的情况就是说你那个消费到了这个消息然后消费者那边自动提交了offset让kafka以为你已经消费好了这个消息其实你刚准备处理这个消息你还没处理你自己就挂了此时这条消息就丢咯。
大家都知道kafka会自动提交offset那么只要关闭自动提交offset在处理完之后自己手动提交offset就可以保证数据不会丢。但是此时确实还是会重复消费比如你刚处理完还没提交offset结果自己挂了此时肯定会重复消费一次自己保证幂等性就好了。
生产环境碰到的一个问题就是说我们的kafka消费者消费到了数据之后是写到一个内存的queue里先缓冲一下结果有的时候你刚把消息写入内存queue然后消费者会自动提交offset。
然后此时我们重启了系统就会导致内存queue里还没来得及处理的数据就丢失了
2kafka弄丢了数据
这块比较常见的一个场景就是kafka某个broker宕机然后重新选举partiton的leader时。大家想想要是此时其他的follower刚好还有些数据没有同步结果此时leader挂了然后选举某个follower成leader之后他不就少了一些数据这就丢了一些数据啊。
所以此时一般是要求起码设置如下4个参数
给这个topic设置replication.factor参数这个值必须大于1要求每个partition必须有至少2个副本。
在kafka服务端设置min.insync.replicas参数这个值必须大于1这个是要求一个leader至少感知到有至少一个follower还跟自己保持联系没掉队这样才能确保leader挂了还有一个follower吧。
在producer端设置acksall这个是要求每条数据必须是写入所有replica之后才能认为是写成功了。
在producer端设置retriesMAX很大很大很大的一个值无限次重试的意思这个是要求一旦写入失败就无限重试卡在这里了。
3生产者会不会弄丢数据
如果按照上述的思路设置了ackall一定不会丢要求是你的leader接收到消息所有的follower都同步到了消息之后才认为本次写成功了。如果没满足这个条件生产者会自动不断的重试重试无限次。
2、如何实现Kafka的高性能
1、宏观架构层面利用Partition实现并行处理
Kafka中每个Topic都包含一个或多个Partition不同Partition可位于不同节点。同时Partition在物理上对应一个本地文件夹每个Partition包含一个或多个Segment每个Segment包含一个数据文件和一个与之对应的索引文件。在逻辑上可以把一个Partition当作一个非常长的数组可通过这个“数组”的索引offset去访问其数据。
一方面由于不同Partition可位于不同机器因此可以充分利用集群优势实现机器间的并行处理。另一方面由于Partition在物理上对应一个文件夹即使多个Partition位于同一个节点也可通过配置让同一节点上的不同Partition置于不同的disk drive上从而实现磁盘间的并行处理充分发挥多磁盘的优势。
利用多磁盘的具体方法是将不同磁盘mount到不同目录然后在server.properties中将log.dirs设置为多目录用逗号分隔。Kafka会自动将所有Partition尽可能均匀分配到不同目录也即不同目录也即不同disk上。
Partition是最小并发粒度Partition个数决定了可能的最大并行度。
2、充分利用PageCache
Page Cache又称pcache其中文名称为页高速缓冲存储器简称页高缓。page cache的大小为一页通常为4K。在linux读写文件时它用于缓存文件的逻辑内容从而加快对磁盘上映像和数据的访问。 是Linux操作系统的一个特色。 1、读Cache
当内核发起一个读请求时(例如进程发起read()请求)首先会检查请求的数据是否缓存到了Page Cache中。
如果有那么直接从内存中读取不需要访问磁盘这被称为cache命中(cache hit)
如果cache中没有请求的数据即cache未命中(cache miss)就必须从磁盘中读取数据。然后内核将读取的数据缓存到cache中这样后续的读请求就可以命中cache了。
page可以只缓存一个文件部分的内容不需要把整个文件都缓存进来。
2、写Cache
当内核发起一个写请求时(例如进程发起write()请求)同样是直接往cache中写入后备存储中的内容不会直接更新(当服务器出现断电关机时存在数据丢失风险)。
内核会将被写入的page标记为dirty并将其加入dirty list中。内核会周期性地将dirty list中的page写回到磁盘上从而使磁盘上的数据和内存中缓存的数据一致。
当满足以下两个条件之一将触发脏数据刷新到磁盘操作
数据存在的时间超过了dirty_expire_centisecs(默认300厘秒即30秒)时间
脏数据所占内存 dirty_background_ratio也就是说当脏数据所占用的内存占总内存的比例超过dirty_background_ratio(默认10即系统内存的10%)的时候会触发pdflush刷新脏数据。
如何查看Page Cache参数
执行命令 sysctl -a|grep dirty
如何调整内核参数来优化IO性能
(1)vm.dirty_background_ratio参数优化
这个参数指定了当文件系统缓存脏页数量达到系统内存百分之多少时如5%就会触发后台回写进程运行将一定缓存的脏页异步地刷入磁盘
当cached中缓存当数据占总内存的比例达到这个参数设定的值时将触发刷磁盘操作。
把这个参数适当调小这样可以把原来一个大的IO刷盘操作变为多个小的IO刷盘操作从而把IO写峰值削平。
对于内存很大和磁盘性能比较差的服务器应该把这个值设置的小一点。
(2)vm.dirty_ratio参数优化
这个参数则指定了当文件系统缓存脏页数量达到系统内存百分之多少时如10%系统不得不开始处理缓存脏页因为此时脏页数量已经比较多为了避免数据丢失需要将一定脏页刷入外存在此过程中很多应用进程可能会因为系统转而处理文件IO而阻塞。
对于写压力特别大的建议把这个参数适当调大对于写压力小的可以适当调小如果cached的数据所占比例(这里是占总内存的比例)超过这个设置
系统会停止所有的应用层的IO写操作等待刷完数据后恢复IO。所以万一触发了系统的这个操作对于用户来说影响非常大的。
(3)vm.dirty_expire_centisecs参数优化
这个参数会和参数vm.dirty_background_ratio一起来作用一个表示大小比例一个表示时间即满足其中任何一个的条件都达到刷盘的条件。
为什么要这么设计呢我们来试想一下以下场景
如果只有参数 vm.dirty_background_ratio 也就是说cache中的数据需要超过这个阀值才会满足刷磁盘的条件
如果数据一直没有达到这个阀值那相当于cache中的数据就永远无法持久化到磁盘这种情况下一旦服务器重启那么cache中的数据必然丢失。
结合以上情况所以添加了一个数据过期时间参数。当数据量没有达到阀值但是达到了我们设定的过期时间同样可以实现数据刷盘。
这样可以有效的解决上述存在的问题其实这种设计在绝大部分框架中都有。
(4)vm.dirty_writeback_centisecs参数优化
理论上调小这个参数可以提高刷磁盘的频率从而尽快把脏数据刷新到磁盘上。但一定要保证间隔时间内一定可以让数据刷盘完成。
(5)vm.swappiness参数优化
禁用swap空间设置vm.swappiness0
3、减少网络开销批处理
批处理是一种常用的用于提高I/O性能的方式。对Kafka而言批处理既减少了网络传输的Overhead又提高了写磁盘的效率。
Kafka 的send方法并非立即将消息发送出去而是通过batch.size和linger.ms控制实际发送频率从而实现批量发送。
由于每次网络传输除了传输消息本身以外还要传输非常多的网络协议本身的一些内容称为Overhead所以将多条消息合并到一起传输可有效减少网络传输的Overhead进而提高了传输效率。
4、数据压缩降低网络负载
Kafka支持将数据压缩后再传输给Broker。除了可以将每条消息单独压缩然后传输外Kafka还支持在批量发送时将整个Batch的消息一起压缩后传输。数据压缩的一个基本原理是重复数据越多压缩效果越好。因此将整个Batch的数据一起压缩能更大幅度减小数据量从而更大程度提高网络传输效率。
Broker接收消息后并不直接解压缩而是直接将消息以压缩后的形式持久化到磁盘。Consumer Fetch到数据后再解压缩。因此Kafka的压缩不仅减少了Producer到Broker的网络传输负载同时也降低了Broker磁盘操作的负载也降低了Consumer与Broker间的网络传输量从而极大得提高了传输效率提高了吞吐量。
5、高效的序列化方式
Kafka消息的Key和Value的类型可自定义只需同时提供相应的序列化器和反序列化器即可。
因此用户可以通过使用快速且紧凑的序列化-反序列化方式如AvroProtocal Buffer来减少实际网络传输和磁盘存储的数据规模从而提高吞吐率。这里要注意如果使用的序列化方法太慢即使压缩比非常高最终的效率也不一定高。
