1.延迟算法应用
2.Kafka源码分析(五) - Server端 - 基于时间轮的通电延时组件
延迟算法应用
延迟算法应用?
1.软件延时
利用多个指令的执行来延时,累加每个指令的延时源码延运行时间,来计算出延时的电路总时间。一般写成一个延时函数。程序
如,通电以下是延时源码延批量预约源码ms软件延时。
delay_ms() {
int c = ; // 调整常数,电路以达到要求的程序延时,但很难!通电
while(c != 0) {
c--;
}
}
以上函数被调用一次,延时源码延就延时ms,电路多次调用可以达到任意更大的程序时间要求。
!通电但是延时源码延,在延时时,电路就其它什么事也做不了了,就是干等啊!
2.硬件延时
利用定时器/计数器芯片,或用微控制器内部的定时器/计数器,实际上,它就是lacp源码对晶振的分频(分频系数可编程设置),得到一个精确的低频的周期信号,用这个周期信号(比如ms)去触发中断,每ms调用一次定时中断服务程序。在定时中断服务程序中加入计数变量,就可以得到任意的定时了。
在ms没有到时,微控制器可以运行其它程序,ms到时再自动进去中断服务程序以处理定时任务,不会像软件延时阻塞了。
3.操作系统中,hll源码都有个硬件延时,和定时中断,可以看ucos ii中的源码,节拍时钟,和汇编语言实现的定时中断。
4.硬件延时,要占用一个定时器/计数器硬件资源。
Kafka源码分析(五) - Server端 - 基于时间轮的延时组件
Kafka内部处理大量的延时操作,例如,在接收到PRODUCE请求后,segmentfault 源码副本可以等待一个timeout的时间再响应客户端。下面我们来探讨一个问题:为什么Kafka要自己实现一个延时任务组件,而不是直接使用Java的java.util.concurrent.DelayQueue呢?我们可以从以下两个方面来分析这个问题。
1.1 DelayQueue的能力
DelayQueue相关的接口/类如下所示:
相应地,DelayQueue提供的能力如下:
1.2 Kafka的业务场景
Kafka的业务背景具有以下特点:
相应地,Kafka对延时任务组件有以下两点要求:
这两点要求都无法通过直接应用DelayQueue的方式得到满足。
二. 组件接口
让我们来看看Kafka的延时任务组件对外提供的接口,从而了解其提供的能力和使用方式。
如下所示:
左边的两个类定义了"延时操作",右边的modelandview源码DelayedOperationPurgatory类定义了一个维护DelayOperaton的容器,其核心操作如下:
三. 实现
以下是关于"延时"实现方式的介绍。
3.1 业务模型
时间轮延时组件的思路如下:
接下来,通过一个具体的例子来说明这种映射逻辑:
首先关注上图中①号时间轮。圆环中的每一个单元格表示一个TimerTaskList。单元格有其关联的时间跨度;下方的"1s x "表示时间轮上共有个单元格,每个单元格的时间跨度为1秒。有一个指针指向了"当前时间"所对应的单元格。顺时针方向为时间流动方向。
当收到一个延迟时间在0-1s的TimerTask时,会将其追加到①号时间轮的橙色单元格中。当收到一个延迟时间在3-4s的TimerTask时,会将其追加到①号时间轮的**单元格中。以此类推。
现在有一个问题:①号时间轮能表示的最大延迟时间是秒,那如果收到了延迟秒的任务该怎么办?这时该用到②号时间轮了,我们称②号为①号的"溢出时间轮"。②号时间轮的特点如下:
如此,延迟时间在-s的TimerTask会被追加到②号的紫色单元格,延迟时间在-s的TimerTask会被追加到②号的绿色单元格中。③号时间轮同理。
刚刚是按①->②->③的顺序来分析时间轮的逻辑,反过来也可以得到有用的想象手里有一个"放大镜",其实③号时间轮的蓝色单元格"放大"后是②号时间轮;②号时间轮的蓝色单元格"放大"后是①号时间轮;蓝色单元格并不实际存储TimerTask。
3.2 数据结构
DelayedOperationPurgatory有一个Timer类型的timeoutTimer属性,用于维护延时任务。实际使用的是Timer的实现类:SystemTimer。该类用于维护延时任务的核心属性有两个:delayQueue和timingWheel。TimingWheel表示单个时间轮,接下来我们来看看其类图:
各属性含义如下:
3.3 算法
3.3.1 添加任务
添加任务的入口是DelayedOperationPurgatory.tryCompleteElseWatch,其核心逻辑分为如下两步:
SystemTimer.add直接调用了addTimerTaskEntry方法,后者逻辑如下:
TimingWheel.add的逻辑也很清晰,分如下4种场景处理:
3.3.2 尝试提前触发任务
入口是DelayedOperationPurgatory.checkAndComplete:
接下来看Watchers.tryCompleteWatched方法的内容:
DelayedOperation.maybeTryComplete方法最终调用了DelayedOperation.tryComplete;
DelayedOperation的子类需要在后者中实现自己的"触发条件"检查逻辑;若满足了提前触发的条件,则调用forceComplete方法执行事件触发场景下的业务逻辑。
3.3.3 任务到期自动执行
DelayedOperationPurgatory中维护了一个expirationReaper线程,其职责就是循环调用kafka.utils.timer.SystemTimer#advanceClock来从时间轮中获取已超时的任务,并更新时间轮的"当前时间"指针。
四. 总结
才疏学浅,未能窥其十之一二,随时欢迎各位交流补充。若文章质量还算及格,可以点赞收藏加以鼓励,后续我继续更新。
另外,也可以在目录中找到同系列的其他文章:
感谢阅读。