原 荐 简单说说Kafka中的时间轮算法
圖片描述(最多50字)
如果你理解了上面的定義，那么就不必往下看了。但如果你第一次看到和我一樣懵比，并且有不少疑問，那么這篇博文將帶你進一步了解時間輪，甚至理解時間輪算法。如果有興趣，可以去看看其他的定時器 你真的了解延時隊列嗎 。博主認為，時間輪定時器最大的優點：是任務的添加與移除，都是O(1)級的復雜度；
不會占用大量的資源；
只需要有一個線程去推進時間輪就可以工作了。
我們將對時間輪做層層推進的解析：
private Task[很長] tasks;
public List<Task> getTaskList(long timestamp) {
return task.get(timestamp)
}
// 假裝這里真的能一毫秒一個循環
public void run(){
while (true){
getTaskList(System.currentTimeMillis()).后臺執行()
Thread.sleep(1);
}
}
假如這個數組長度達到了億億級，我們確實可以這么干。 那如果將精度縮減到秒級呢？我們也需要一個百億級長度的數組。
/ 一個精度為秒級的延時任務管理類 /
private Map<Long, Task> taskMap;
public List<Task> getTaskList(long timestamp) {
return taskMap.get(timestamp - timestamp % 1000)
}
// 新增一個任務
public void addTask(long timestamp, Task task) {
List<Task> taskList = getTaskList(timestamp - timestamp % 1000);
if (taskList == null){
taskList = new ArrayList();
}
taskList.add(task);
}
// 假裝這里真的能一秒一個循環
public void run(){
while (true){
getTaskList(System.currentTimeMillis()).后臺執行()
Thread.sleep(1000);
}
}
其實時間輪就是一個不存在hash沖突的數據結構
圖片描述(最多50字)
就拿秒表來說，它總是落在 0 - 59 秒，每走一圈，又會重新開始。用偽代碼模擬一下我們這個秒表：private Bucket[60] buckets;// 表示60秒
public void addTask(long timestamp, Task task) {
Bucket bucket = buckets[timestamp / 1000 % 60];
bucket.add(task);
}
public Bucket getBucket(long timestamp) {
return buckets[timestamp / 1000 % 60];
}
// 假裝這里真的能一秒一個循環
public void run(){
while (true){
getBucket(System.currentTimeMillis()).后臺執行()
Thread.sleep(1000);
}
}
這樣，我們的時間總能落在0 - 59任意一個bucket上，就如同我們的秒鐘總是落在0 - 59刻度上一樣，這便是 時間輪的環形隊列 。
public class TimeWheel {
/ 一個時間槽的時間 */
private long tickMs;
/* 時間輪大小 /
private int wheelSize;
/ 時間跨度 */
private long interval;
/ 槽 */
private Bucket[] buckets;
/* 時間輪指針 /
private long currentTimestamp;
/ 上層時間輪 /
private volatile TimeWheel overflowWheel;
public TimeWheel(long tickMs, int wheelSize, long currentTimestamp) {
this.currentTimestamp = currentTimestamp;
this.tickMs = tickMs;
this.wheelSize = wheelSize;
this.interval = tickMs wheelSize;
this.buckets = new Bucket[wheelSize];
this.currentTimestamp = currentTimestamp - (currentTimestamp % tickMs);
for (int i = 0; i < wheelSize; i++) {
buckets[i] = new Bucket();
}
}
}
將任務添加到時間輪中十分簡單，對于每個時間輪來說，比如說秒級時間輪，和分級時間輪，都有它自己的過期槽。也就是delayMs < tickMs的時候。
1）比如說有一個任務要在 16:29:07 執行，從秒級時間輪中來看，當我們的當前時間走到16:29:06的時候，則表示這個任務已經過期了。因為它的delayMs = 1000ms，小于了我們的秒級時間輪的tickMs(1000ms)。
比如說有一個任務要在 16:41:25 執行，從分級時間輪中來看，當我們的當前時間走到 16:41的時候（ 分級時間輪沒有秒針！它的最小精度是分鐘（一定要理解這一點） ），則表示這個任務已經到期，因為它的delayMs = 25000ms，小于了我們的分級時間輪的tickMs（60000ms）。
二、時間未到期，且delayMs小于interval。
/**
- 添加任務到某個時間輪
*/
public boolean addTask(TimedTask timedTask) {
long expireTimestamp = timedTask.getExpireTimestamp();
long delayMs = expireTimestamp - currentTimestamp;
if (delayMs < tickMs) {// 到期了
return false;
} else {
// 扔進當前時間輪的某個槽中，只有時間【大于某個槽】，才會放進去
if (delayMs < interval) {
int bucketIndex = (int) (((delayMs + currentTimestamp) / tickMs) % wheelSize);
Bucket bucket = buckets[bucketIndex];
bucket.addTask(timedTask);
} else {
// 當maybeInThisBucket大于等于wheelSize時，需要將它扔到上一層的時間輪
TimeWheel timeWheel = getOverflowWheel();
timeWheel.addTask(timedTask);
}
}
return true;
}
/** -
獲取或創建一個上層時間輪
*/
private TimeWheel getOverflowWheel() {
if (overflowWheel == null) {
synchronized (this) {
if (overflowWheel == null) {
overflowWheel = new TimeWheel(interval, wheelSize, currentTimestamp, delayQueue);
}
}
}
return overflowWheel;
}
當然我們的時間輪還需要一個指針的推進機制，總不能讓時間永遠停留在當前吧？推進的時候，同時類似遞歸，去推進一下上一層的時間輪。注意：要強調一點的是，我們這個時間輪更像是電子表，它不存在時間的中間狀態，也就是精度這個概念一定要理解好。比如說，對于秒級時間輪來說，它的精度只能保證到1秒，小于1秒的，都會當成是已到期
對于分級時間輪來說，它的精度只能保證到1分，小于1分的，都會當成是已到期
/**
-
嘗試推進一下指針
*/
public void advanceClock(long timestamp) {
if (timestamp >= currentTimestamp + tickMs) {
currentTimestamp = timestamp - (timestamp % tickMs);
if (overflowWheel != null) {
this.getOverflowWheel()
.advanceClock(timestamp);
}
}
}
三、對于高層時間輪來說，精度越來越不準，會不會有影響？上面說到，分級時間輪，精度只有分鐘級，總不能延遲1秒的定時任務和延遲59秒的定時任務同時執行吧？
有這個疑問的同學很好！實際上很好解決，只需再入時間輪即可。比如說，對于分鐘級時間輪來說，delayMs為1秒和delayMs為59秒的都已經過期，我們將其取出，再扔進底層的時間輪不就可以了？
1秒的會被扔到秒級時間輪的下一個執行槽中，而59秒的會被扔到秒級時間輪的后59個時間槽中。
細心的同學會發現，我們的添加任務方法，返回的是一個bool
public boolean addTask(TimedTask timedTask)
再倒回去好好看看，添加到最底層時間輪失敗的（我們只能直接操作最底層的時間輪，不能直接操作上層的時間輪），是不是會直接返回flase？ 對于再入失敗的任務，我們直接執行即可。
/**
-
將任務添加到時間輪
*/
public void addOrSubmitTask(TimedTask timedTask) {
if (!timeWheel.addTask(timedTask)) {
taskExecutor.submit(timedTask.getTask());
}
}
四、如何知道一個任務已經過期？記得我們將任務存儲在槽中嘛？比如說秒級時間輪中，有60個槽，那么一共有60個槽。如果時間輪共有兩層，也僅僅只有120個槽。我們只需將槽扔進一個delayedQueue之中即可。
我們輪詢地從delayedQueue取出已經過期的槽即可。（前面的所有代碼，為了簡單說明，并沒有引入這個DelayQueue的概念，所以不用去上面翻了，并沒有。博主覺得... 已經看到這里了，應該很明白這個DelayQueue的意義了。 ）
其實簡單來說，實際上定時任務單單使用DelayQueue來實現，也是可以的，但是一旦任務的數量多了起來，達到了百萬級，千萬級，針對這個delayQueue的增刪，將非常的慢。
一、面向槽的delayQueue
而對于時間輪來說，它只需要往delayQueue里面扔各種槽即可，比如我們的定時任務長短不一，最長的跨度到了24年，這個delayQueue也僅僅只有300個元素。
二、處理過期的槽
而這個槽到期后，也就是被我們從delayQueue中poll出來后，我們只需要將槽中的所有任務循環一次，重新加到新的槽中（添加失敗則直接執行）即可。
/**
- 推進一下時間輪的指針，并且將delayQueue中的任務取出來再重新扔進去
*/
public void advanceClock(long timeout) {
try {
Bucket bucket = delayQueue.poll(timeout, TimeUnit.MILLISECONDS);
if (bucket != null) {
timeWheel.advanceClock(bucket.getExpire());
bucket.flush(this::addTask);
}
} catch (Exception e) {
e.printStackTrace();
}
}
轉載于:https://blog.51cto.com/14028890/2309569
與50位技術專家面對面20年技術見證，附贈技術全景圖總結
以上是生活随笔為你收集整理的原 荐 简单说说Kafka中的时间轮算法的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 在 sql server 中，查询 数据
- 下一篇： js --- 递归结构图