【演算法資料結構專題】「延時佇列演算法」史上手把手教你針對層級時間輪（TimingWheel）實作延時佇列的開發實戰落地（下）-有解無憂

承接上文

承接上一篇文章【演算法資料結構專題】「延時佇列演算法」史上手把手教你針對層級時間輪（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，任何足夠先進的科技，都與魔法無異，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/549501.html

標籤：其他

上一篇：java -- 例外處理、Collection、Iterator迭代器、泛型

下一篇：Rust編程語言入門之函式式語言特性：-迭代器和閉包