Kafka 延迟队列_kafka延时队列实现

Survive by day and develop by night.
talk for import biz , show your perfect code,full busy，skip hardness,make a better result,wait for change,challenge Survive.
happy for hardess to solve denpendies.

目录

概述

kafka 延迟队列

需求：

kafka延迟队列

设计思路

实现思路分析

1.延迟队列

1.简介
TimingWheel 是kafka 时间轮的实现，内部包含了一个TimerTaskList 数组，每个数组包含了一些链表组成的TimerTaskEntry事件，每个TimerTaskList表示时间轮的某一格，这一格的时间跨度为tickMs，同⼀个TimerTaskList中的事件都是相差在⼀个tickMs跨度内的，整个时间轮的时间跨度为interval = tickMs * wheelSize，该时间轮能处理的时间范围在cuurentTime到currentTime + interval之间的事件。
当添加⼀个时间他的超时时间⼤于整个时间轮的跨度时， expiration >= currentTime + interval，则会将该事件向上级传递，上级的tickMs是下级的interval，传递直到某⼀个时间轮满⾜expiration < currentTime + interval，

然后计算对应位于哪⼀格，然后将事件放进去，重新设置超时时间，然后放进jdk延迟队列
SystemTimer会取出queue中的TimerTaskList，根据expiration将currentTime往前推进，然后把⾥⾯所有的事件重新放进时间轮中，因为ct推进了，所以有些事件会在第0格，表示到期了，直接返回。

else if (expiration < currentTime + tickMs) {

然后将任务提交到java线程池中处理。
服务端在处理客户端的请求，针对不同的请求，可能不会⽴即返回响应结果给客户端。在处理这类请求时，服务端会为这类请求创建延迟操作对象放⼊延迟缓存队列中。延迟缓存的数据结构类似MAP，延迟操作对象从延迟缓存队列中完成并移除有两种⽅式：

延迟操作对应的外部事件发⽣时，外部事件会尝试完成延迟缓存中的延迟操作 。
如果外部事件仍然没有完成延迟操作，超时时间达到后，会强制完成延迟的操作 。

2. 延时操作接口

DelayedOperation​​接⼝表示延迟的操作对象。此接⼝的实现类包括延迟加⼊，延迟⼼跳，延迟⽣产，延迟拉取。延迟接⼝相关的⽅法：
​​forceComplete​​​：强制完成，两个地⽅调⽤，尝试完成⽅法（​​tryComplete​​）返回true时；延迟操作超时时。

​​run​​​：线程运⾏，延迟操作超时后，会调⽤线程的运⾏⽅法，只会调⽤⼀次，因为超时就会发⽣⼀次。超时后会调⽤强制完成⽅法（​​forceComplete​​），如果返回true，会调⽤超时的回调⽅法。

​​onComplete​​：完成的回调⽅法。

​​onExpiration​​：超时的回调⽅法。

外部事件触发完成和超时完成都会调⽤​​forceComplete()​​​，并调⽤​​onComplete()​​​。​​forceComplete​​​和​​onComplete​​​只会调⽤⼀次。多线程下⽤原⼦变量来控制只有⼀个线程会调⽤​​onComplete​​​和​​forceComplete​​。

延迟⽣产和延迟拉取完成时的回调⽅法，尝试完成的延迟操作

副本管理器在创建延迟操作时，会把回调⽅法传给延迟操作对象。当延迟操作完成时，在​​onComplete​​⽅法中会调⽤回调⽅法，返回响应结果给客户端。

拓展实现

这里参考：github:简单实现上述流程：
入门级实现:
: 部分源码实现.
: 源码实现

性能参数测试：

每秒大概18-20个请求，主要用于网络IO操作耗费了不少时间。

参考资料和推荐阅读

1. 爬虫框架的设计与实现之JAVA篇.
2. 主流爬虫框架的基本介绍.
3. 高拓展性的Java多线程爬虫框架reptile.

欢迎阅读，各位老铁，如果对你有帮助，点个赞加个关注呗！~