【演算法數據結構專題】「延時隊列演算法」史上手把手教你針對層級時間輪（TimingWheel）實現延時隊列的開發實戰落地（下）

承接上文承接上一篇文章【演算法數據結構專題】「延時隊列演算法」史上手把手教你針對層級時間輪（TimingWheel）實現延時隊列的開發實戰落地（上）】我們基本上對層級時間輪演算法的基本原理有了一定的認識，本章節就從落地的角度進行分析和介紹如何通過Java進行實現一個屬於我們自己的時間輪服務組件，最後，在 ...

承接上文

承接上一篇文章【演算法數據結構專題】「延時隊列演算法」史上手把手教你針對層級時間輪（TimingWheel）實現延時隊列的開發實戰落地（上）】我們基本上對層級時間輪演算法的基本原理有了一定的認識，本章節就從落地的角度進行分析和介紹如何通過Java進行實現一個屬於我們自己的時間輪服務組件，最後，在告訴大家一下，其實時間輪的技術是來源於生活中的時鐘。

時間輪演示結構總覽

無序列表時間輪

【無序列表時間輪】主要是由LinkedList鏈表和啟動線程、終止線程實現。

遍歷定時器中所有節點，將剩餘時間為 0s 的任務進行過期處理，在執行一個周期。

無序鏈表：每一個延時任務都存儲在該鏈表當中（無序存儲）。
啟動線程：直接在鏈表後面push ，時間複雜度 O(1)。
終止線程：直接在鏈表中刪除節點，時間複雜度 O(1) 。

遍歷周期：需要遍歷鏈表中所有節點，時間複雜度 O(n)，所以伴隨著鏈表中的元素越來越多，速度也會越來越慢！

無序列表時間輪的長度限制了其適用場景，這裡對此進行優化。因此引入了有序列表時間輪。

有序列表時間輪

與無序列表時間輪一樣，同樣使用鏈表進行實現和設計，但存儲的是絕對延時時間點。

啟動線程：有序插入，比較時間按照時間大小有序插入，時間複雜度O(n)，主要耗時在插入操作。
終止線程：鏈表中查找任務，刪除節點，時間複雜度O(n)，主要耗時在插入操作。

找到執行最後一個過期任務即可，無需遍歷整個鏈表，時間複雜度 O(1)，從上面的描述「有序列表定時器」的性能瓶頸在於插入時的任務排序，但是換來的就是縮短了遍歷周期。

所以我們如果要提高性，就必須要提升一下插入和刪除以及檢索的性能，因此引入了「樹形有序列表時間輪」在「有序列表定時器」的基礎上進行優化，以有序樹的形式進行任務存儲。

樹形有序列表時間輪

啟動定時器：有序插入，比較時間按照時間大小有序插入，時間複雜度 O(logn)
終止定時器：在鏈表中查找任務，刪除節點，時間複雜度 O(logn)
周期清算: 找到執行最後一個過期任務即可，無需遍歷整個鏈表，時間複雜度 O(1)

層級時間輪

整體流程架構圖，如下所示。

對應的原理，在這裡就不進行贅述了，之前本人已經有兩篇文章對層級式時間輪進行了較為詳細的介紹了，有需要的小伙伴，可以直接去前幾篇文章去學習，接下來我們進行相關的實現。

時間輪數據模型

時間輪（TimingWheel）是一個存儲定時任務的環形隊列，數組中的每個元素可以存放一個定時任務列表，其中存放了真正的定時任務，如下圖所示。

時間輪的最基本邏輯模型，由多個時間格組成，每個時間格代表當前時間輪的基本時間跨度（tickMs），所以我們先來設計和定義開發對應的時間輪的輪盤模型。命名為Roulette類。

輪盤抽象類-Roulette

之所以定義這個抽象類

public abstract class Roulette {
	// 鏈表數據-主要用於存儲每個延時任務節點
    List<TimewheelTask> tasks = null;
    // 游標指針索引
    protected int index;
	// 時間輪輪盤的容量大小，如果是分鐘級別，一般就是60個格
    protected int capacity;
	// 時間輪輪盤的層級，如果是一級，它的上級就是二級
    protected Integer level;
    private AtomicInteger num = new AtomicInteger(0);

   // 構造器
    public Roulette(int capacity, Integer level) {
        this.capacity = capacity;
        this.level = level;
        this.tasks = new ArrayList<>(capacity);
        this.index = 0;
    }
    // 獲取當前下表的索引對應的時間輪的任務
    public TimewheelTask getTask() {
        return tasks.get(index);
    }
   // init初始化操作機制
    public List<TimewheelTask> init() {
        long interval = MathTool.power((capacity + 1), level);
        long add = 0;
        TimewheelTask delayTask = null;
        for (int i = 0; i < capacity; i++) {
            add += interval;
            if (level == 0) {
                delayTask = new DefaultDelayTask(level);
            } else {
                delayTask = new SplitDelayTask(level);
            }
            //已經轉換為最小的時間間隔
            delayTask.setDelay(add, TimeUnitProvider.getTimeUnit());
            tasks.add(delayTask);
        }
        return tasks;
    }

   // 索引下標移動
    public void indexAdd() {
        this.index++;
        if (this.index >= capacity) {
            this.index = 0;
        }
    }
   // 添加對應的任務到對應的隊列裡面
    public void addTask(TimewheelTask task) {
        tasks.add(task);
    }
	// 給子類提供的方法進行實現對應的任務添加功能
    public abstract void addTask(int interval, MyTask task);
}

時間輪盤的熟悉信息介紹

鏈表數據-主要用於存儲每個延時任務節點。

List<TimewheelTask> tasks = null;

tasks也可以改成雙向鏈表 + 數組的結構：即節點存貯的對象中有指針，組成環形，可以通過數組的下標靈活訪問每個節點，類似 LinkedHashMap。

游標指針索引

protected int index;

時間輪輪盤的容量大小，如果是分鐘級別，一般就是60個格

protected int capacity;

時間輪輪盤的層級，如果是一級，它的上級就是二級

protected Integer level;

init初始化時間輪輪盤對象模型，主要用於分配分配每一個輪盤上面元素的TimewheelTask，用於延時隊列的執行任務線程，已經分配對應的每一個節點的延時時間節點數據。

 public List<TimewheelTask> init() {
	   //  那麼整個時間輪的總體時間跨度（interval）
        long interval = MathTool.power((capacity + 1), level);
        long add = 0;
        TimewheelTask delayTask = null;
        for (int i = 0; i < capacity; i++) {
            add += interval;
            if (level == 0) {
                delayTask = new ExecuteTimewheelTask(level);
            } else {
                delayTask = new MoveTimewheelTask(level);
            }
            //已經轉換為最小的時間間隔
            delayTask.setDelay(add, TimeUnitProvider.getTimeUnit());
            tasks.add(delayTask);
        }
        return tasks;
}

整數a的n次冪：interval，計算跨度，主要是各級別之間屬於平方倍數

例如，第一層：20 ，第二層：20^2 ......

    //例如 n=7  二進位 0   1                 1          1
    //a的n次冪 = a的2次冪×a的2次冪  ×   a的1次冪×a的1次冪  ×a
    public static long power(long a, int n) {
        int rtn = 1;
        while (n >= 1) {
            if((n & 1) == 1){
                rtn *= a;
            }
            a *= a;
            n = n >> 1;
        }
        return rtn;
    }

TimeUnitProvider工具類

主要用於計算時間單位操作的轉換

public class TimeUnitProvider {
    private static TimeUnit unit = TimeUnit.SECONDS;
    public static TimeUnit getTimeUnit() {
        return unit;
    }
}

代碼簡介：

interval：代表著初始化的延時時間數據值，主要用於不同的層次的出發時間數據
for (int i = 0; i < capacity; i++) ：代表著進行for迴圈進行添加對應的延時隊列任務到集合中
add += interval，主要用於添加對應的延時隊列的延時數據值！並且分配給當前輪盤得到所有數據節點。

獲取當前下標的索引對應的時間輪的任務節點

public TimewheelTask getTask() {
        return tasks.get(index);
}

層級時間輪的Bucket數據桶

在這裡我們建立了一個TimewheelBucket類實現了Roulette輪盤模型，從而進行建立對應的我們的層級時間輪的數據模型，並且覆蓋了addTask方法。

public class TimewheelBucket extends Roulette {

    public TimewheelBucket(int capacity, Integer level) {
        super(capacity, level);
    }

    public synchronized void addTask(int interval, MyTask task) {
        interval -= 1;
        int curIndex = interval + this.index;
        if (curIndex >= capacity) {
            curIndex = curIndex - capacity;
        }
        tasks.get(curIndex).addTask(task);
    }
}

添加addTask方法，進行獲取計算對應的下標，並且此方法add操作才是對外開發調用的，在這裡，我們主要實現了根據層級計算出對應的下標進行獲取對應的任務執行調度點，將我們外界BizTask，真正的業務操作封裝到這個BizTask模型，交由我們的系統框架進行執行。

     public synchronized void addTask(int interval, BizTask task) {
        interval -= 1;
        int curIndex = interval + this.index;
        if (curIndex >= capacity) {
            curIndex = curIndex - capacity;
        }
        tasks.get(curIndex).addTask(task);
    }

時間輪輪盤上的任務點

我們針對於時間輪輪盤的任務點進行設計和定義對應的調度執行任務模型。一個調度任務點，可以幫到關係到多個BizTask，也就是用戶提交上來的業務任務線程對象，為了方便採用延時隊列的延時處理模式，再次實現了Delayed這個介面，對應的實現代碼如下所示：

Delayed介面

public interface Delayed extends Comparable<Delayed> {
    /**
     * Returns the remaining delay associated with this object, in the
     * given time unit.
     *
     * @param unit the time unit
     * @return the remaining delay; zero or negative values indicate
     * that the delay has already elapsed
     */
    long getDelay(TimeUnit unit);
}

TimewheelTask時間輪刻度點

@Getter
public abstract class TimewheelTask implements Delayed {
    private List<BizTask> tasks = new ArrayList<BizTask>();
    private int level;
    private Long delay;
    private long calDelay;
    private TimeUnit calUnit;
    public TimewheelTask(int level) {
        this.level = level;
    }
    public void setDelay(Long delay, TimeUnit unit) {
        this.calDelay=delay;
        this.calUnit=unit;
    }
    public void calDelay() {
        this.delay = TimeUnit.NANOSECONDS.convert(this.calDelay, this.calUnit) + System.nanoTime(); 
    }
    public long getDelay(TimeUnit unit) {
        return this.delay - System.nanoTime();
    }
    public int compareTo(Delayed o) {
        long d = (getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS));
        return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
    }
    public void addTask(BizTask task) {
        synchronized (this) {
            tasks.add(task);
        }
    }
    public void clear() {
        tasks.clear();
    }
    public abstract void run();
}

業務任務集合：private List<BizTask> tasks = new ArrayList<BizTask>();
- 層級
```
private int level;
```
- 延時時間
```
  private Long delay;
```
- 實際用於延時計算的時間，就是底層是統一化所有的延時時間到對應的延時隊列
```
private long calDelay;
```
- 實際用於延時計算的時間，就是底層是統一化所有的延時時間到對應的延時隊列（用於統一化的時間單位）
```
private TimeUnit calUnit;
```

添加對應的業務延時任務到輪盤刻度點

    public void addTask(BizTask task) {
        synchronized (this) {
            tasks.add(task);
        }
    }

刻度點的實現類

因為對應的任務可能會需要將下游的業務任務進行升級或者降級，所以我們會針對於執行任務點分為，執行任務刻度點和躍遷任務刻度點兩種類型。

執行任務延時隊列刻度點

public class ExecuteTimewheelTask extends TimewheelTask {
    public ExecuteTimewheelTask(int level) {
        super(level);
    }
    //到時間執行所有的任務
    public void run() {
        List<BizTask> tasks = getTasks();
        if (CollectionUtils.isNotEmpty(tasks)) {
            tasks.forEach(task -> ThreadPool.submit(task));
        }
    }
}

再次我們就定義執行這些任務的線程池為：

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 100, 3, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(10000),
            new MyThreadFactory("executor"), new ThreadPoolExecutor.CallerRunsPolicy());

躍遷任務延時隊列刻度點

public class MoveTimewheelTask extends TimewheelTask {
    public MoveTimewheelTask(int level) {
        super(level);
    }
    //躍遷到其他輪盤，將對應的任務
    public void run() {
        List<BizTask> tasks = getTasks();
        if (CollectionUtils.isNotEmpty(tasks)) {
            tasks.forEach(task -> {
                long delay = task.getDelay();
                TimerWheel.adddTask(task,delay, TimeUnitProvider.getTimeUnit());
            });
        }
    }
}

致辭整個時間輪輪盤的數據模型就定義的差不多了，接下來我們需要定義運行在時間輪盤上面的任務模型，BizTask基礎模型。

BizTask基礎模型

public abstract class BizTask implements Runnable {
     protected long interval;
     protected int index;
     protected long executeTime;
     public BizTask(long interval, TimeUnit unit, int index) {
          this.interval  = interval;
          this.index = index;
          this.executeTime= TimeUnitProvider.getTimeUnit().convert(interval,unit)+TimeUnitProvider.getTimeUnit().convert(System.nanoTime(),TimeUnit.NANOSECONDS);
     }
     public long getDelay() {
          return this.executeTime - TimeUnitProvider.getTimeUnit().convert(System.nanoTime(), TimeUnit.NANOSECONDS);
     }
}

主要針對於任務執行，需要交給線程池去執行，故此，實現了Runnable介面。

protected long interval;：跨度操作
protected int index;：索引下表，在整個隊列裡面的下表處理
protected long executeTime;：對應的執行時間

其中最重要的便是獲取延時時間的操作，主要提供給框架的Delayed介面進行判斷是否到執行時間了。

     public long getDelay() {
          return this.executeTime - TimeUnitProvider.getTimeUnit().convert(System.nanoTime(), TimeUnit.NANOSECONDS);
     }

層級時間輪的門面TimerWheel

最後我們要進行定義和設計開發對應的整體的時間輪層級模型。

public class TimerWheel {

    private static Map<Integer, TimewheelBucket> cache = new ConcurrentHashMap<>();
    //一個輪表示三十秒
    private static int interval = 30;
    private static wheelThread wheelThread;

    public static void adddTask(BizTask task, Long time, TimeUnit unit) {
        if(task == null){
            return;
        }
        long intervalTime = TimeUnitProvider.getTimeUnit().convert(time, unit);
        if(intervalTime < 1){
            ThreadPool.submit(task);
            return;
        }
        Integer[] wheel = getWheel(intervalTime,interval);
        TimewheelBucket taskList = cache.get(wheel[0]);
        if (taskList != null) {
            taskList.addTask(wheel[1], task);
        } else {
            synchronized (cache) {
                if (cache.get(wheel[0]) == null) {
                    taskList = new TimewheelBucket(interval-1, wheel[0]);
                    wheelThread.add(taskList.init());
                    cache.putIfAbsent(wheel[0],taskList);
                }
            }
            taskList.addTask(wheel[1], task);
        }
    }
    static{
        interval = 30;
        wheelThread = new wheelThread();
        wheelThread.setDaemon(false);
        wheelThread.start();
    }
    private static Integer[] getWheel(long intervalTime,long baseInterval) {
        //轉換後的延時時間
        if (intervalTime < baseInterval) {
            return new Integer[]{0, Integer.valueOf(String.valueOf((intervalTime % 30)))};
        } else {
            return getWheel(intervalTime,baseInterval,baseInterval, 1);
        }
    }

    private static Integer[] getWheel(long intervalTime,long baseInterval,long interval, int p) {
         long nextInterval = baseInterval * interval;
        if (intervalTime < nextInterval) {
            return new Integer[]{p, Integer.valueOf(String.valueOf(intervalTime / interval))};
        } else {
            return getWheel(intervalTime,baseInterval,nextInterval, (p+1));
        }
    }

    static class wheelThread extends Thread {
        DelayQueue<TimewheelTask> queue = new DelayQueue<TimewheelTask>();

        public DelayQueue<TimewheelTask> getQueue() {
            return queue;
        }

        public void add(List<TimewheelTask> tasks) {
            if (CollectionUtils.isNotEmpty(tasks)) {
                tasks.forEach(task -> add(task));
            }
        }

        public void add(TimewheelTask task) {
            task.calDelay();
            queue.add(task);
        }

        @Override
        public void run() {
            while (true) {
                try {
                    TimewheelTask task = queue.take();
                    int p = task.getLevel();
                    long nextInterval = MathTool.power(interval, Integer.valueOf(String.valueOf(MathTool.power(2, p))));
                    TimewheelBucket timewheelBucket = cache.get(p);
                    synchronized (timewheelBucket) {
                        timewheelBucket.indexAdd();
                        task.run();
                        task.clear();
                    }
                    task.setDelay(nextInterval, TimeUnitProvider.getTimeUnit());
                    task.calDelay();
                    queue.add(task);
                } catch (InterruptedException e) {

                }
            }
        }
    }
}

TimerWheel的模型定義

private static Map<Integer, TimewheelBucket> cache = new ConcurrentHashMap<>();

一個輪表示30秒的整體跨度。

private static int interval = 30;

創建整體驅動的執行線程

private static wheelThread wheelThread;

 static{
        interval = 30;
        wheelThread = new wheelThread();
        wheelThread.setDaemon(false);
        wheelThread.start();
}

    static class wheelThread extends Thread {
        DelayQueue<TimewheelTask> queue = new DelayQueue<TimewheelTask>();
        public DelayQueue<TimewheelTask> getQueue() {
            return queue;
        }
        public void add(List<TimewheelTask> tasks) {
            if (CollectionUtils.isNotEmpty(tasks)) {
                tasks.forEach(task -> add(task));
            }
        }
        public void add(TimewheelTask task) {
            task.calDelay();
            queue.add(task);
        }
        @Override
        public void run() {
            while (true) {
                try {
                    TimewheelTask task = queue.take();
                    int p = task.getLevel();
                    long nextInterval = MathTool.power(interval, Integer.valueOf(String.valueOf(MathTool.power(2, p))));
                    TimewheelBucket timewheelBucket = cache.get(p);
                    synchronized (timewheelBucket) {
                        timewheelBucket.indexAdd();
                        task.run();
                        task.clear();
                    }
                    task.setDelay(nextInterval, TimeUnitProvider.getTimeUnit());
                    task.calDelay();
                    queue.add(task);
                } catch (InterruptedException e) {

                }
            }
   }

獲取對應的時間輪輪盤模型體系

    private static Integer[] getWheel(long intervalTime,long baseInterval) {
        //轉換後的延時時間
        if (intervalTime < baseInterval) {
            return new Integer[]{0, Integer.valueOf(String.valueOf((intervalTime % 30)))};
        } else {
            return getWheel(intervalTime,baseInterval,baseInterval, 1);
        }
    }

    private static Integer[] getWheel(long intervalTime,long baseInterval,long interval, int p) {
         long nextInterval = baseInterval * interval;
        if (intervalTime < nextInterval) {
            return new Integer[]{p, Integer.valueOf(String.valueOf(intervalTime / interval))};
        } else {
            return getWheel(intervalTime,baseInterval,nextInterval, (p+1));
        }
    }

到這裡相信大家，基本上應該是瞭解瞭如何去實現對應的時間輪盤的技術實現過程，有興趣希望整個完整源碼的，可以聯繫我哦。謝謝大家!

本文來自博客園，作者：洛神灬殤，轉載請註明原文鏈接：https://www.cnblogs.com/liboware/p/17299146.html，任何足夠先進的科技，都與魔法無異。