概述

延時任務(wù)相信大家都不陌生，在現(xiàn)實(shí)的業(yè)務(wù)中應(yīng)用場景可以說是比比皆是。例如訂單下單15分鐘未支付直接取消，外賣超時自動賠付等等。這些情況下，我們該怎么設(shè)計我們的服務(wù)的實(shí)現(xiàn)呢？

笨一點(diǎn)的方法自然是定時任務(wù)去數(shù)據(jù)庫進(jìn)行輪詢。但是當(dāng)業(yè)務(wù)量較大，事件處理比較費(fèi)時的時候，我們的系統(tǒng)和數(shù)據(jù)庫往往會面臨巨大的壓力，如果采用這種方式或許會導(dǎo)致數(shù)據(jù)庫和系統(tǒng)的崩潰。那么有什么好辦法嗎？今天我來為大家介紹幾種實(shí)現(xiàn)延時任務(wù)的辦法。

JAVA DelayQueue

你沒看錯，java內(nèi)部有內(nèi)置延時隊列，位于java concurrent包內(nèi)。

DelayQueue是一個jdk中自帶的延時隊列實(shí)現(xiàn)，他的實(shí)現(xiàn)依賴于可重入鎖ReentrantLock以及條件鎖Condition和優(yōu)先隊列PriorityQueue。而且本質(zhì)上他也是一個阻塞隊列。那么他是如何實(shí)現(xiàn)延時效果的呢。

DelayQueue的實(shí)現(xiàn)原理

首先DelayQueue隊列中的元素必須繼承一個接口叫做Delayed，我們找到這個類

public interface Delayed extends Comparable<Delayed> {
        long getDelay(TimeUnit unit);
    }

發(fā)現(xiàn)這個類內(nèi)部定義了一個返回值為long的方法getDelay，這個方法用來定義隊列中的元素的過期時間，所有需要放在隊列中的元素，必須實(shí)現(xiàn)這個方法。

然后我們來看看延遲隊列的隊列是如何操作的，我們就拿最典型的offer和take來看：

public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

offer操作平平無奇，甚至直接調(diào)用到了優(yōu)先隊列的offer來將隊列根據(jù)延時進(jìn)行排序，只不過加了個鎖，做了些數(shù)據(jù)的調(diào)整，沒有什么深入的地方，但是take的實(shí)現(xiàn)看上去就很復(fù)雜了。（注意，Dalayed繼承了Comparable方法，所以是可以直接用優(yōu)先隊列來排序的，只要你自己實(shí)現(xiàn)了compareTo方法）我嘗試加了些注釋讓各位看得更明白些：

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 自選操作
            for (;;) {
                // 獲取隊列第一個元素，如果隊列為空
                // 阻塞住直到有新元素加入隊列，offer等方法調(diào)用signal喚醒線程
                E first = q.peek();
                if (first == null)
                    available.await();
                else {
                    // 如果隊列中有元素
                    long delay = first.getDelay(NANOSECONDS);
                    // 判斷延時時間，如果到時間了，直接取出數(shù)據(jù)并return
                    if (delay <= 0)
                        return q.poll();
                    first = null;
                    // 如果leader為空則阻塞
                    if (leader != null)
                        available.await();
                    else {
                        // 獲取當(dāng)前線程
                        Thread thisThread = Thread.currentThread();
                        // 設(shè)置leader為當(dāng)前線程
                        leader = thisThread;
                        try {
                            // 阻塞延時時間
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

我們可以看到take的實(shí)現(xiàn)依靠了無限自旋，直到第一個隊列元素過了超時時間后才會返回，否則等待他的只有被阻塞。

DelayQueue實(shí)現(xiàn)延時隊列的優(yōu)缺點(diǎn)

看了源碼后，我們應(yīng)該對DelayQueue的實(shí)現(xiàn)有了一個大致的了解，也對他的優(yōu)缺點(diǎn)有了一定的理解。他的優(yōu)點(diǎn)很明顯：

• java原生支持，不需要引入第三方工具
• 線程安全，即插即用使用方便

但是他的缺點(diǎn)也是很明顯的：

• 不支持分布式，并且數(shù)據(jù)放在內(nèi)存中，沒有持久化的支持，服務(wù)宕機(jī)會丟失數(shù)據(jù)
• 插入時使用的是優(yōu)先隊列的排序，時間復(fù)雜度較高，并且對于隊列中的任務(wù)不能很好的管理

所以有沒有更好的延時隊列的實(shí)現(xiàn)呢，我們繼續(xù)看下去~

時間輪算法

時間輪算法是一個被設(shè)計出來處理延時任務(wù)的算法，現(xiàn)實(shí)中的應(yīng)用可以在kafka以及netty等項目中找到類似的實(shí)現(xiàn)。

時間輪的具體實(shí)現(xiàn)

所謂時間輪，顧名思義，他是一個類似于時鐘的結(jié)構(gòu)，即他的主結(jié)構(gòu)是一個環(huán)形數(shù)組，如圖：

環(huán)形數(shù)組中存放的是一個一個的鏈表，鏈表中存放著需要執(zhí)行的任務(wù)，我們設(shè)定好數(shù)組中執(zhí)行的間隔，假設(shè)我們的環(huán)形數(shù)組的長度是60，每個數(shù)組的執(zhí)行間隔為1s，那么我們會在每過1s就會執(zhí)行數(shù)組下一個元素中的鏈表中的元素。如果只是這樣，那么我們將無法處理60秒之外的延時任務(wù)，這顯然不合適，所以我們會在每個任務(wù)中加上一個參數(shù)圈數(shù)，來表明任務(wù)會在幾圈后執(zhí)行。假如我們有一個任務(wù)是在150s后執(zhí)行，那么他應(yīng)該在30s的位置，同時圈數(shù)應(yīng)該為2。我們每次執(zhí)行一個鏈表中的任務(wù)的時候會把當(dāng)圈需要執(zhí)行的任務(wù)取出執(zhí)行，然后把他從鏈表中刪除，如果任務(wù)不是當(dāng)圈執(zhí)行，則修改他的圈數(shù)，將圈數(shù)減1，于是一個簡單的時間輪出爐了。

那么這樣的時間輪有什么優(yōu)缺點(diǎn)呢？

先來說優(yōu)點(diǎn)吧：

• 相比DelayQueue來說，時間輪的插入更加的高效，時間復(fù)雜度為O(1)
• 實(shí)現(xiàn)簡單清晰，任務(wù)調(diào)度更加方便合理

當(dāng)然他的缺點(diǎn)也不少：

• 他和DelayQueue一樣不支持分布式，并且數(shù)據(jù)放在內(nèi)存中，沒有持久化的支持，服務(wù)宕機(jī)會丟失數(shù)據(jù)
• 數(shù)組間的間隔設(shè)置會影響任務(wù)的精度
• 由于不同圈數(shù)的任務(wù)會在同一個鏈表中，執(zhí)行到每個數(shù)組元素時需要遍歷所有的鏈表數(shù)據(jù)，效率會很低

進(jìn)階優(yōu)化版時間輪算法

剛才提到了一些時間輪算法的缺點(diǎn)，那么是不是有一些方法來進(jìn)行下優(yōu)化？這里我來介紹一下時間輪的優(yōu)化版本。

之前我們提到不同圈數(shù)的任務(wù)會在同一個鏈表中被重復(fù)遍歷影響效率，這種情況下我們可以進(jìn)行如下優(yōu)化：將時間輪進(jìn)行分層

我們可以看到圖中，我們采用了多層級的設(shè)計，上圖中分了三層，每層都是60格，第一個輪盤中的間隔為1小時，我們的數(shù)據(jù)每一次都是插入到這個輪盤中，每當(dāng)這個輪盤經(jīng)過一個小時后來到下一個刻度，就會取出其中的所有元素，按照延遲時間放入到第二個象征著分鐘的輪盤中，以此類推。

這樣的實(shí)現(xiàn)好處可以說是顯而易見的：

• 首先避免了當(dāng)時間跨度較大時空間的浪費(fèi)
• 每一次到達(dá)刻度的時候我們不用再像以前那樣遍歷鏈表取出需要的數(shù)據(jù)，而是可以一次性全部拿出來，大大節(jié)約了操作的時間

時間輪算法的應(yīng)用

時間輪算法可能在之前大家沒有聽說過，但是他在各個地方都有著不小的作用。linux的定時器的實(shí)現(xiàn)中就有時間輪的身影，同樣如果你是一個喜好看源碼的讀者，你也可能會在kafka以及netty中找到他的實(shí)現(xiàn)。

kafka

kafka中應(yīng)用了時間輪算法，他的實(shí)現(xiàn)和之前提到的進(jìn)階版時間輪沒有太大的區(qū)別，只有在一點(diǎn)上：kafka內(nèi)部實(shí)現(xiàn)的時間輪應(yīng)用到了DelayQueue。

@nonthreadsafe
    private[timer] class TimingWheel(tickMs: Long, wheelSize: Int, startMs: Long, taskCounter: AtomicInteger, queue: DelayQueue[TimerTaskList]) {

    private[this] val interval = tickMs * wheelSize
    private[this] val buckets = Array.tabulate[TimerTaskList](wheelSize) { _ => new TimerTaskList(taskCounter) }

    private[this] var currentTime = startMs - (startMs % tickMs)
    
    @volatile private[this] var overflowWheel: TimingWheel = null

    private[this] def addOverflowWheel(): Unit = {
        synchronized {
        if (overflowWheel == null) {
            overflowWheel = new TimingWheel(
            tickMs = interval,
            wheelSize = wheelSize,
            startMs = currentTime,
            taskCounter = taskCounter,
            queue
            )
        }
        }
    }

    def add(timerTaskEntry: TimerTaskEntry): Boolean = {
        val expiration = timerTaskEntry.expirationMs

        if (timerTaskEntry.cancelled) {
        false
        } else if (expiration < currentTime + tickMs) {
        false
        } else if (expiration < currentTime + interval) {
        val virtualId = expiration / tickMs
        val bucket = buckets((virtualId % wheelSize.toLong).toInt)
        bucket.add(timerTaskEntry)

        if (bucket.setExpiration(virtualId * tickMs)) {
            queue.offer(bucket)
        }
        true
        } else {
        if (overflowWheel == null) addOverflowWheel()
        overflowWheel.add(timerTaskEntry)
        }
    }

    def advanceClock(timeMs: Long): Unit = {
        if (timeMs >= currentTime + tickMs) {
        currentTime = timeMs - (timeMs % tickMs)

        if (overflowWheel != null) overflowWheel.advanceClock(currentTime)
        }
    }
    }

上面是kafka內(nèi)部的實(shí)現(xiàn)（使用的語言是scala），我們可以看到實(shí)現(xiàn)非常的簡潔，并且使用到了DelayQueue。我們剛才已經(jīng)討論過了DelayQueue的優(yōu)缺點(diǎn)，查看源碼后我們已經(jīng)可以有一個大致的結(jié)論了：DelayQueue在kafka的時間輪中的作用是負(fù)責(zé)推進(jìn)任務(wù)的，為的就是防止在時間輪中由于任務(wù)比較稀疏而造成的"空推進(jìn)"。DelayQueue的觸發(fā)機(jī)制可以很好的避免這一點(diǎn)，同時由于DelayQueue的插入效率較低，所以僅用于底層的推進(jìn)，任務(wù)的插入由時間輪來操作，兩者配置，可以實(shí)現(xiàn)效率和資源的平衡。

netty

netty的內(nèi)部也有時間輪的實(shí)現(xiàn)HashedWheelTimer

HashedWheelTimer的實(shí)現(xiàn)要比kafka內(nèi)部的實(shí)現(xiàn)復(fù)雜許多，和kafka不同的是，它的內(nèi)部推進(jìn)不是依靠的DelayQueue而是自己實(shí)現(xiàn)了一套，源碼太長，有興趣的讀者可以自己去看一下。

小結(jié)

時間輪說了這么多，我們可以看到他的效率是很出眾的，但是還是有這么一個問題：他不支持分布式。當(dāng)我們的業(yè)務(wù)很復(fù)雜，需要分布式的時候，時間輪顯得力不從心，那么這個時候有什么好一點(diǎn)的延時隊列的選擇呢？我們或許可以嘗試使用第三方的工具

redis延時隊列

其實(shí)啊說起延時，我們?nèi)绻Ｓ?code>redis的話，就會想起redis是存在過期機(jī)制的，那么我們是否可以利用這個機(jī)制來實(shí)現(xiàn)一個延時隊列呢？

redis自帶key的過期機(jī)制，而且可以設(shè)置過期后的回調(diào)方法。基于此特性，我們可以非常容易就完成一個延時隊列，任務(wù)進(jìn)來時，設(shè)定定時時間，并且配置好過期回調(diào)方法即可。

除了使用redis的過期機(jī)制之外，我們也可以利用它自帶的zset來實(shí)現(xiàn)延時隊列。zset支持高性能的排序，因此我們?nèi)蝿?wù)進(jìn)來時可以將時間戳作為排序的依據(jù)，以此將任務(wù)的執(zhí)行先后進(jìn)行有序的排列，這樣也能實(shí)現(xiàn)延時隊列。

zset實(shí)現(xiàn)延時隊列的好處：

• 支持高性能排序
• redis本身的高可用和高性能以及持久性

mq延時隊列

rocketmq延時消息

rocketmq天然支持延時消息，他的延時消息分為18個等級，每個等級對應(yīng)不同的延時時間。

那么他的原理是怎樣的呢？

rocketmq的broker收到消息后會將消息寫入commitlog，并且判斷這個消息是否是延時消息（即delay屬性是否大于0），之后如果判斷確實(shí)是延時消息，那么他不會馬上寫入，而是通過轉(zhuǎn)發(fā)的方式將消息放入對應(yīng)的延時topic（18個延時級別對應(yīng)18個topic）

rocketmq會有一個定時任務(wù)進(jìn)行輪詢，如果任務(wù)的延遲時間已經(jīng)到了就發(fā)往指定的topic。

這個設(shè)計比較的簡單粗暴，但是缺點(diǎn)也十分明顯：

• 延時是固定的，如果想要的延遲超出18個級別就沒辦法實(shí)現(xiàn)
• 無法實(shí)現(xiàn)精準(zhǔn)延時，隊列的堆積等等情況也會導(dǎo)致執(zhí)行產(chǎn)生誤差

rocketmq的精準(zhǔn)延時消息

rocketmq本身是不支持的精確延遲的，他的商業(yè)版本ons倒是支持。不過rocketmq的社區(qū)中有相應(yīng)的解決方案。方案是借助于時間輪算法來實(shí)現(xiàn)的，感興趣的朋友可以自行去社區(qū)查看。（社區(qū)中的一些未被合并的pr是不錯的實(shí)現(xiàn)參考）

總結(jié)

延時隊列的實(shí)現(xiàn)千千萬，但是如果要在生產(chǎn)中大規(guī)模使用，那么大部分情況下其實(shí)都避不開時間輪算法。改進(jìn)過的時間輪算法可以做到精準(zhǔn)延時，持久化，高性能，高可用性，可謂是完美。但是話又說回來，其他的延時方式就無用了嗎？其實(shí)不是的，所有的方式都是需要匹配自己的使用場景。如果你是極少量數(shù)據(jù)的輪詢，那么定時輪詢數(shù)據(jù)庫或許才是最佳的解決方案，而不是無腦的引入復(fù)雜的延時隊列。如果是單機(jī)的任務(wù)，那么jdk的延時隊列也是不錯的選擇。

本文介紹的這些延時隊列只是為了向大家展示他們的原理和優(yōu)缺點(diǎn)，具體的使用還需要結(jié)合自己業(yè)務(wù)的場景。

以上就是一文帶你深入了解Java中延時任務(wù)的實(shí)現(xiàn)的詳細(xì)內(nèi)容，更多關(guān)于Java延時任務(wù)的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論