定時任務(wù)之HashedWheelTimer

Redisson分布式鎖的實現(xiàn)一文中抑月,我們說到Redisson會調(diào)用scheduleExpirationRenewal方法創(chuàng)建一個定時任務(wù)來刷新鎖的過期時間,防止任務(wù)執(zhí)行完畢前鎖就過期釋放了秒咨。在那篇文章中舌涨,我們沒有詳述這個定時任務(wù)的原理蜘犁,本文中我們來探究一番定時任務(wù)的原理团赁。

定時任務(wù)的本質(zhì)

一個Timer本質(zhì)上是這樣的一個數(shù)據(jù)結(jié)構(gòu):deadline越近的任務(wù)擁有越高優(yōu)先級育拨,提供以下3中基本操作:

  1. schedule新增
  2. cancel刪除
  3. expire執(zhí)行到期的任務(wù)
  4. updateDeadline更行到期時間(可選)

expire通常有兩種工作方式:

  1. 輪詢

    每隔一個時間片就去查找哪些任務(wù)已經(jīng)到期

  2. 睡眠/喚醒

    1. 不停地查找deadline最近的任務(wù),如到期則執(zhí)行欢摄;否則sleep直到其到期
    2. 在sleep期間熬丧,如果有任務(wù)被cancelschedule,則deadline最近的任務(wù)有可能改變怀挠,線程會被喚醒并重新進(jìn)行1的邏輯

具體實現(xiàn)的數(shù)據(jù)結(jié)構(gòu)可以有很多選擇:(假設(shè)任務(wù)持有自己在總體任務(wù)集合中對應(yīng)的節(jié)點析蝴,cancel時不需要查找的過程)

  1. 有序鏈表

    • schedule:O(n)
    • cancel:O(1) //雙向鏈表的節(jié)點刪除
    • expire:O(1) //不停地查看第一個就可以了
  2. 堆heap

    • schedule:O(log2N) //調(diào)整heap
    • cancel:O(log2N) //調(diào)整heap
    • expire:O(1)

HashedWheelTimer

Redisson使用的定時任務(wù)是Netty提供的HashedWheelTimer

image.png

Hash Wheel Timer是一個環(huán)形結(jié)構(gòu)绿淋,可以想象成時鐘闷畸,分為很多格子,一個格子代表一段時間(越短Timer精度越高)吞滞,并用一個List保存在該格子上到期的所有任務(wù)佑菩。同時一個指針隨著時間流逝一格一格轉(zhuǎn)動,并執(zhí)行對應(yīng)List中所有到期的任務(wù)裁赠。任務(wù)通過取模決定應(yīng)該放入哪個格子殿漠。

以上圖為例,假設(shè)一個格子是1秒佩捞,則整個wheel能表示的時間段為8s绞幌,假設(shè)當(dāng)前指針指向2,此時需要調(diào)度一個3s后執(zhí)行的任務(wù)一忱,顯然應(yīng)該加入到(2+3=5)的方格中啊奄,指針再走3次就可以執(zhí)行了;如果任務(wù)要在10s后執(zhí)行掀潮,應(yīng)該等指針走完一個round零2格再執(zhí)行菇夸,因此應(yīng)放入4,同時將round(1)保存到任務(wù)中仪吧。檢查到期任務(wù)應(yīng)當(dāng)只執(zhí)行round為0的庄新,格子上其他任務(wù)的round應(yīng)減1.

  • schedule:O(1)
  • cancel:O(1)
  • expire:最壞情況O(n),平均O(1) //顯然格子越多每個格子對應(yīng)的List就越短薯鼠,越接近O(1)择诈;最壞情況下所有的任務(wù)都在一個格子中,O(n)

使用

HashedWheelTimer的使用如下所示:

@Test
public void test01() throws IOException {
    HashedWheelTimer timer = new HashedWheelTimer();    //使用默認(rèn)參數(shù)
    logger.info("start");
    timer.newTimeout(timeout -> logger.info("running"), 3, TimeUnit.SECONDS);   //延時3秒后開始執(zhí)行

    System.in.read();
}

執(zhí)行結(jié)果如下:

14:16:50.743 [main] INFO love.wangqi.timer.TimerTest - start
14:16:53.888 [pool-1-thread-1] INFO love.wangqi.timer.TimerTest - running

從日志的打印時間出皇,我們看到程序運(yùn)行的大概3秒鐘之后羞芍,我們定義的TimerTask開始執(zhí)行。

我們知道郊艘,HashedWheelTimer以一個固定的時間間隔前進(jìn)一個格子荷科,并且激活對應(yīng)格子里面的任務(wù)唯咬,但是這里有個缺陷,就是任務(wù)是串行的畏浆,也就是所有的任務(wù)是依次執(zhí)行胆胰,如果調(diào)度的任務(wù)耗時比較長的話,容易出現(xiàn)調(diào)度超時的情況刻获,因此很可能產(chǎn)生任務(wù)堆集的情況出現(xiàn)蜀涨。

以如下代碼所示:


@Test
public void test02() throws IOException {
    HashedWheelTimer timer = new HashedWheelTimer();
    logger.info("start");
    timer.newTimeout(timeout -> {
        logger.info("running");
        Thread.sleep(2000);
        logger.info("end");
    }, 1, TimeUnit.SECONDS);

    timer.newTimeout(timeout -> {
        logger.info("running");
        Thread.sleep(2000);
        logger.info("end");
    }, 1, TimeUnit.SECONDS);

    System.in.read();
}

執(zhí)行結(jié)果如下:

14:24:02.872 [main] INFO love.wangqi.timer.TimerTest - start
14:24:04.020 [pool-1-thread-1] INFO love.wangqi.timer.TimerTest - running
14:24:06.023 [pool-1-thread-1] INFO love.wangqi.timer.TimerTest - end
14:24:06.023 [pool-1-thread-1] INFO love.wangqi.timer.TimerTest - running
14:24:08.025 [pool-1-thread-1] INFO love.wangqi.timer.TimerTest - end

我們的本意其實是程序執(zhí)行后延時1秒,然后兩個任務(wù)同時開始執(zhí)行蝎毡。但實際的執(zhí)行結(jié)果顯示task2task1阻塞了厚柳,直到task1執(zhí)行結(jié)束task2才開始執(zhí)行。這就導(dǎo)致了串行化的任務(wù)調(diào)度延時沐兵,因此草娜,應(yīng)該避免耗時比較長的任務(wù)在同一個線程中執(zhí)行。

源碼解讀

關(guān)鍵的成員變量

// 指針轉(zhuǎn)動以及任務(wù)執(zhí)行的線程
private final Worker worker = new Worker();
private final Thread workerThread;

public static final int WORKER_STATE_INIT = 0;
public static final int WORKER_STATE_STARTED = 1;
public static final int WORKER_STATE_SHUTDOWN = 2;
// 工作線程的狀態(tài)
private volatile int workerState;

// 每個tick所需的時間
private final long tickDuration;

// 時間輪的結(jié)構(gòu)
private final HashedWheelBucket[] wheel;

構(gòu)造方法


public HashedWheelTimer(
        ThreadFactory threadFactory,    // 用于創(chuàng)建worker線程的線程工廠
        long tickDuration,              // tick的時長痒筒,也就是指針多久轉(zhuǎn)一格
        TimeUnit unit,                  // tickDuration的時間單位
        int ticksPerWheel,              // 一圈有幾個格子
        boolean leakDetection,          // 是否開啟內(nèi)存泄露檢測
        long maxPendingTimeouts         // 任務(wù)的超時等待時間
        ) {
    // 一些參數(shù)校驗
    if (threadFactory == null) {
        throw new NullPointerException("threadFactory");
    }
    if (unit == null) {
        throw new NullPointerException("unit");
    }
    if (tickDuration <= 0) {
        throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
    }
    if (ticksPerWheel <= 0) {
        throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
    }

    // 創(chuàng)建時間輪基本的數(shù)據(jù)結(jié)構(gòu),一個數(shù)組茬贵。長度為不小于ticksPerWheel的最小2的n次方
    wheel = createWheel(ticksPerWheel);
    // 掩碼簿透,用于計算某個任務(wù)落在哪個格子上。
    // 因為wheel.length為2的n次方解藻,mask = 2^n - 1低位將全部是1老充,所以任務(wù)的deadline&mast == 任務(wù)的deadline&wheel.length
    mask = wheel.length - 1;

    // 轉(zhuǎn)換成納秒
    this.tickDuration = unit.toNanos(tickDuration);

    // 檢驗是否存在溢出。即指針轉(zhuǎn)動的時間間隔不能太長而導(dǎo)致tickDuration*wheel.length>Long.MAX_VALUE
    if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
        throw new IllegalArgumentException(String.format(
                "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
                tickDuration, Long.MAX_VALUE / wheel.length));
    }
    // 創(chuàng)建worker線程
    workerThread = threadFactory.newThread(worker);
    // 這里默認(rèn)是啟動內(nèi)存泄露檢測:當(dāng)HashedWheelTimer實例超過當(dāng)前CPU可用核數(shù)*4的時候螟左,將發(fā)出警告
    leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
    // 最大的任務(wù)數(shù)量啡浊。當(dāng)HashedWheelTimer實例上的任務(wù)超出這個數(shù)量時會拋出錯誤
    this.maxPendingTimeouts = maxPendingTimeouts;

    // 如果HashedWheelTimer實例數(shù)超過了64,會報警胶背。
    // 因為時間輪是一個非常消耗資源的結(jié)構(gòu)所以實例數(shù)目不能太高
    if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
        WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
        reportTooManyInstances();
    }
}

添加定時任務(wù)

在前面的實例中巷嚣,我們看到使用newTimeout方法來添加定時任務(wù)。newTimeout方法的代碼如下:


@Override
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
    // 參數(shù)校驗
    if (task == null) {
        throw new NullPointerException("task");
    }
    if (unit == null) {
        throw new NullPointerException("unit");
    }
    // 增加任務(wù)數(shù)量
    long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
    // 如果設(shè)置了最大任務(wù)數(shù)钳吟,且任務(wù)數(shù)量超過最大任務(wù)數(shù)廷粒,拋出異常
    if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
        pendingTimeouts.decrementAndGet();
        throw new RejectedExecutionException("Number of pending timeouts ("
            + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
            + "timeouts (" + maxPendingTimeouts + ")");
    }
    // 如果沒有啟動時間輪,則啟動
    start();

    // Add the timeout to the timeout queue which will be processed on the next tick.
    // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
    // 計算任務(wù)的deadline
    long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;

    // Guard against overflow.
    if (delay > 0 && deadline < 0) {
        deadline = Long.MAX_VALUE;
    }
    // 這里定時任務(wù)不是直接加到對應(yīng)的格子中红且,而是先加入到一個隊列里坝茎,然后等到下一個tick的時候,會從隊列里取出最多100000個任務(wù)加入到指定的格子中
    HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
    timeouts.add(timeout);
    return timeout;
}

這里使用的Queue不是普通Java自帶的Queue的實現(xiàn)暇番,而是使用JCTool——一個高性能的并發(fā)Queue實現(xiàn)包嗤放。

啟動時間輪


// 這個方法不需要用戶顯式地調(diào)用。因為在添加定時任務(wù)(newTimeout方法)的時候會自動調(diào)用此方法
// 如果沒有定時任務(wù)壁酬,時間輪就不需要啟動
public void start() {
    // 判斷當(dāng)前時間輪的狀態(tài)次酌。
    // 如果是初始化狀態(tài)恨课,則采用CAS的方法將狀態(tài)修改為開始狀態(tài),然后啟動時間輪線程和措。
    switch (WORKER_STATE_UPDATER.get(this)) {
        case WORKER_STATE_INIT:
            if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                workerThread.start();
            }
            break;
        case WORKER_STATE_STARTED:
            break;
        case WORKER_STATE_SHUTDOWN:
            throw new IllegalStateException("cannot be started once stopped");
        default:
            throw new Error("Invalid WorkerState");
    }

    // Wait until the startTime is initialized by the worker.
    // 等待woker線程初始化時間輪的啟動時間
    while (startTime == 0) {
        try {
            startTimeInitialized.await();
        } catch (InterruptedException ignore) {
            // Ignore - it will be ready very soon.
        }
    }
}

AtomicIntegerFieldUpdater是JUC里面的類庄呈,原理是利用安全的反射進(jìn)行原子操作,來獲取實例的本身的屬性派阱。有比AtomicInteger更好的性能和更低的內(nèi)存占用诬留。

停止時間輪

@Override
public Set<Timeout> stop() {
    // worker線程不能停止時間輪,也就是加入的定時任務(wù)不能調(diào)用這個方法贫母。不然會有惡意的定時任務(wù)調(diào)用這個方法而造成大量定時任務(wù)失效
    if (Thread.currentThread() == workerThread) {
        throw new IllegalStateException(
                HashedWheelTimer.class.getSimpleName() +
                        ".stop() cannot be called from " +
                        TimerTask.class.getSimpleName());
    }
    
    // CAS的方式嘗試將當(dāng)前狀態(tài)替換為WORKER_STATE_SHUTDOWN文兑。
    // 如果替換失敗,則當(dāng)前狀態(tài)只能是WORKER_STATE_STARTED或者WORKER_STATE_SHUTDOWN腺劣。直接將當(dāng)前狀態(tài)設(shè)置為WORKER_STATE_SHUTDOWN
    if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
        // workerState can be 0 or 2 at this moment - let it always be 2.
        if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
            INSTANCE_COUNTER.decrementAndGet();
            if (leak != null) {
                boolean closed = leak.close(this);
                assert closed;
            }
        }

        return Collections.emptySet();
    }

    try {
        // 中斷worker線程绿贞,嘗試把正在進(jìn)行任務(wù)的線程中斷掉。
        // 如果某些任務(wù)正在執(zhí)行則會拋出InterruptedException異常橘原,并且任務(wù)會嘗試立即中斷
        boolean interrupted = false;
        while (workerThread.isAlive()) {
            workerThread.interrupt();
            try {
                workerThread.join(100);
            } catch (InterruptedException ignored) {
                interrupted = true;
            }
        }
        // 如果中斷掉了所有工作的線程籍铁,那么當(dāng)前時間輪調(diào)度的線程會在隨后關(guān)閉
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    } finally {
        INSTANCE_COUNTER.decrementAndGet();
        if (leak != null) {
            boolean closed = leak.close(this);
            assert closed;
        }
    }
    // 返回未處理的任務(wù)
    return worker.unprocessedTimeouts();
}

HashedWheelTimeout

newTimeout方法中會將我們的TimerTask包裝成一個HashedWheelTimeout對象,然后添加到Queue<HashedWheelTimeout> timeouts隊列中趾断。

HashedWheelTimeout是一個定時任務(wù)的內(nèi)部包裝類拒名,雙向鏈表結(jié)構(gòu)。保存定時任務(wù)到期執(zhí)行的任務(wù)芋酌、deadline增显、round等信息。

private static final class HashedWheelTimeout implements Timeout {
    // 定義定時任務(wù)的3個狀態(tài):初始化脐帝、取消同云、過期
    private static final int ST_INIT = 0;
    private static final int ST_CANCELLED = 1;
    private static final int ST_EXPIRED = 2;
    // CAS方式更新定時任務(wù)狀態(tài)
    private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");

    // 時間輪引用
    private final HashedWheelTimer timer;
    // 到期需要執(zhí)行的任務(wù)
    private final TimerTask task;
    private final long deadline;

    @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
    private volatile int state = ST_INIT;

    // 離任務(wù)執(zhí)行的輪數(shù),當(dāng)將任務(wù)加入到格子中時計算該值堵腹,每過一輪炸站,該值減1
    long remainingRounds;

    // 雙向鏈表結(jié)構(gòu),由于只有worker線程會訪問疚顷,這里不需要synchronization/volatile
    HashedWheelTimeout next;
    HashedWheelTimeout prev;

    // 定時任務(wù)所在的格子
    HashedWheelBucket bucket;

    HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
        this.timer = timer;
        this.task = task;
        this.deadline = deadline;
    }

    @Override
    public Timer timer() {
        return timer;
    }

    @Override
    public TimerTask task() {
        return task;
    }

    @Override
    public boolean cancel() {
        // 這里只是修改狀態(tài)為ST_CANCELLED武契,會在下次tick時從格子中移除
        if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
            return false;
        }
        // 加入到時間輪的待取消隊列,并在每次tick的時候荡含,從相應(yīng)格子中移除
        timer.cancelledTimeouts.add(this);
        return true;
    }

    // 從格子中移除自身
    void remove() {
        HashedWheelBucket bucket = this.bucket;
        if (bucket != null) {
            bucket.remove(this);
        } else {
            timer.pendingTimeouts.decrementAndGet();
        }
    }

    public boolean compareAndSetState(int expected, int state) {
        return STATE_UPDATER.compareAndSet(this, expected, state);
    }

    public int state() {
        return state;
    }

    @Override
    public boolean isCancelled() {
        return state() == ST_CANCELLED;
    }

    @Override
    public boolean isExpired() {
        return state() == ST_EXPIRED;
    }
    
    // 過期并執(zhí)行任務(wù)
    public void expire() {
        if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
            return;
        }

        try {
            task.run(this);
        } catch (Throwable t) {
            if (logger.isWarnEnabled()) {
                logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
            }
        }
    }
}

HashedWheelBucket

HashedWheelBucket用來存放HashedWheelTimeout咒唆,結(jié)構(gòu)類似于LinkedList。提供了expireTimeouts方法來執(zhí)行格子中的過期任務(wù)


private static final class HashedWheelBucket {
    // 指向格子中任務(wù)的首尾
    private HashedWheelTimeout head;
    private HashedWheelTimeout tail;

    // 添加到鏈表的末尾
    public void addTimeout(HashedWheelTimeout timeout) {
        assert timeout.bucket == null;
        timeout.bucket = this;
        if (head == null) {
            head = tail = timeout;
        } else {
            tail.next = timeout;
            timeout.prev = tail;
            tail = timeout;
        }
    }

    // 執(zhí)行格子中的過期任務(wù)释液,tick到該格子的時候全释,worker線程會調(diào)用這個方法
    // 根據(jù)deadline和remainingRounds判斷任務(wù)是否過期
    public void expireTimeouts(long deadline) {
        HashedWheelTimeout timeout = head;

        // 遍歷格子中的所有定時任務(wù)
        while (timeout != null) {
            HashedWheelTimeout next = timeout.next;
            if (timeout.remainingRounds <= 0) {     // 定時任務(wù)到期
                next = remove(timeout);             // 從鏈表中移除到期任務(wù)
                if (timeout.deadline <= deadline) {
                    timeout.expire();               // 執(zhí)行到期的任務(wù)
                } else {
                    // The timeout was placed into a wrong slot. This should never happen.
                    throw new IllegalStateException(String.format(
                            "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                }
            } else if (timeout.isCancelled()) {     // 如果任務(wù)被取消,從鏈表中移除
                next = remove(timeout);
            } else {
                timeout.remainingRounds --;         // 沒有到期误债,輪數(shù)-1
            }
            timeout = next;
        }
    }

    // 從鏈表中移除到期任務(wù)
    public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
        HashedWheelTimeout next = timeout.next;
        // remove timeout that was either processed or cancelled by updating the linked-list
        if (timeout.prev != null) {
            timeout.prev.next = next;
        }
        if (timeout.next != null) {
            timeout.next.prev = timeout.prev;
        }

        if (timeout == head) {
            // if timeout is also the tail we need to adjust the entry too
            if (timeout == tail) {
                tail = null;
                head = null;
            } else {
                head = next;
            }
        } else if (timeout == tail) {
            // if the timeout is the tail modify the tail to be the prev node.
            tail = timeout.prev;
        }
        // null out prev, next and bucket to allow for GC.
        timeout.prev = null;
        timeout.next = null;
        timeout.bucket = null;
        timeout.timer.pendingTimeouts.decrementAndGet();
        return next;
    }

    // 清空這個格子浸船,將沒有過期或者是已經(jīng)取消的任務(wù)保存在set中
    public void clearTimeouts(Set<Timeout> set) {
        for (;;) {
            HashedWheelTimeout timeout = pollTimeout();
            if (timeout == null) {
                return;
            }
            if (timeout.isExpired() || timeout.isCancelled()) {
                continue;
            }
            set.add(timeout);
        }
    }

    // 從鏈表中取得最頭上的任務(wù)
    private HashedWheelTimeout pollTimeout() {
        HashedWheelTimeout head = this.head;
        if (head == null) {
            return null;
        }
        HashedWheelTimeout next = head.next;
        if (next == null) {
            tail = this.head =  null;
        } else {
            this.head = next;
            next.prev = null;
        }

        // null out prev and next to allow for GC.
        head.next = null;
        head.prev = null;
        head.bucket = null;
        return head;
    }
}

Worker

前面我們看到時間輪啟動時啟動Worker線程妄迁。Worker是時間輪的核心線程類。tick的轉(zhuǎn)動李命,過期任務(wù)的處理都是在這個線程中處理的登淘。

private final class Worker implements Runnable {
    private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();

    private long tick;

    @Override
    public void run() {
        // 初始化startTime,所有任務(wù)的deadline都是相對于這個時間點的
        startTime = System.nanoTime();
        // 由于System.nanoTime()可能返回0封字,甚至負(fù)數(shù)黔州。0是一個標(biāo)示符,用來判斷startTime是否被初始化阔籽,所以當(dāng)startTime=0的時候流妻,重新賦值為1
        if (startTime == 0) {
            startTime = 1;
        }

        // 喚醒阻塞在start()的線程
        startTimeInitialized.countDown();
        
        // 只要時間輪的狀態(tài)為WORKER_STATE_STARTED,就循環(huán)地轉(zhuǎn)動tick笆制,判斷響應(yīng)格子中的到期任務(wù)
        do {
            // waitForNextTick方法主要是計算下次tick的時間绅这,然后sleep到下次tick
            // 返回值是System.nanoTime() - startTime,也就是Timer啟動后到這次tick所過去的時間
            final long deadline = waitForNextTick();
            if (deadline > 0) { // 溢出或者被中斷的時候會返回負(fù)數(shù)在辆,所以小于等于0不管
                // 獲取tick對應(yīng)的格子索引
                int idx = (int) (tick & mask);
                // 移除被取消的任務(wù)
                processCancelledTasks();
                HashedWheelBucket bucket = wheel[idx];
                // 從任務(wù)隊列中取出任務(wù)加入到對應(yīng)的格子中
                transferTimeoutsToBuckets();
                // 執(zhí)行格子中的任務(wù)
                bucket.expireTimeouts(deadline);
                tick++;
            }
        } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);

        // 時間輪停止了证薇。
        // 清除所有格子中的任務(wù),并加入到未處理任務(wù)列表匆篓,以供stop()方法返回
        for (HashedWheelBucket bucket: wheel) {
            bucket.clearTimeouts(unprocessedTimeouts);
        }
        // 將還沒有加入到格子中的待處理定時任務(wù)隊列中的任務(wù)取出浑度,如果是未取消任務(wù),則加入到未處理任務(wù)列表奕删,以供stop()方法返回
        for (;;) {
            HashedWheelTimeout timeout = timeouts.poll();
            if (timeout == null) {
                break;
            }
            if (!timeout.isCancelled()) {
                unprocessedTimeouts.add(timeout);
            }
        }
        // 處理取消的任務(wù)
        processCancelledTasks();
    }

    // 將newTimeout()方法中加入到待處理定時任務(wù)隊列中的任務(wù)加入到指定的格子中
    private void transferTimeoutsToBuckets() {
        // 每次tick只處理10w個任務(wù),以免阻塞worker線程
        for (int i = 0; i < 100000; i++) {
            HashedWheelTimeout timeout = timeouts.poll();
            // 如果沒有任務(wù)了疗认,直接跳出循環(huán)
            if (timeout == null) {
                // all processed
                break;
            }
            // 還沒有放入到格子中就取消了完残,直接略過
            if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
                // Was cancelled in the meantime.
                continue;
            }
            
            // 計算任務(wù)需要經(jīng)過多少個tick
            long calculated = timeout.deadline / tickDuration;
            // 計算任務(wù)的輪數(shù)
            timeout.remainingRounds = (calculated - tick) / wheel.length;

            // 如果任務(wù)在timeouts隊列中放久了,以至于已經(jīng)過了執(zhí)行時間横漏,這個時候就使用當(dāng)前tick谨设,也就是放到當(dāng)前bucket中。此方法調(diào)用完后就會被執(zhí)行缎浇。
            final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
            int stopIndex = (int) (ticks & mask);

            // 將任務(wù)加入到相應(yīng)的格子中
            HashedWheelBucket bucket = wheel[stopIndex];
            bucket.addTimeout(timeout);
        }
    }

    // 將取消的任務(wù)取出扎拣,并從格子中移除
    private void processCancelledTasks() {
        for (;;) {
            HashedWheelTimeout timeout = cancelledTimeouts.poll();
            if (timeout == null) {
                // all processed
                break;
            }
            try {
                timeout.remove();
            } catch (Throwable t) {
                if (logger.isWarnEnabled()) {
                    logger.warn("An exception was thrown while process a cancellation task", t);
                }
            }
        }
    }

    // sleep,直到下次tick到來素跺,然后返回該次tick和啟動時間之間的時長
    private long waitForNextTick() {
        // 下次tick的時間點二蓝,用于計算需要sleep的時間
        long deadline = tickDuration * (tick + 1);

        for (;;) {
            // 計算需要sleep的時間
            final long currentTime = System.nanoTime() - startTime;
            long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;

            if (sleepTimeMs <= 0) {
                if (currentTime == Long.MIN_VALUE) {
                    return -Long.MAX_VALUE;
                } else {
                    return currentTime;
                }
            }

            // 因為windows平臺的定時調(diào)度最小單位為10ms,如果不是10ms的倍數(shù)指厌,可能會引起sleep時間不準(zhǔn)確
            if (PlatformDependent.isWindows()) {
                sleepTimeMs = sleepTimeMs / 10 * 10;
            }

            try {
                Thread.sleep(sleepTimeMs);
            } catch (InterruptedException ignored) {
                // 調(diào)用HashedWheelTimer.stop()時優(yōu)雅退出
                if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
                    return Long.MIN_VALUE;
                }
            }
        }
    }

    public Set<Timeout> unprocessedTimeouts() {
        return Collections.unmodifiableSet(unprocessedTimeouts);
    }
}

總結(jié)

總體來說刊愚,HashedWheelTimer使用的是一個比較樸素的算法,要點有兩個:

  1. 添加定時任務(wù)

    1. 如果worker線程沒有執(zhí)行則啟動worker線程踩验。
    2. 將定時任務(wù)task包裝成HashedWheelTimeout鸥诽,然后添加到Queue<HashedWheelTimeout> timeouts隊列中
  2. worker線程的執(zhí)行

    1. 調(diào)用waitForNextTick方法等待直到下一個tick
    2. 調(diào)用processCancelledTasks方法處理被取消的任務(wù)商玫。從Queue<HashedWheelTimeout> cancelledTimeouts隊列(調(diào)用cancel方法取消任務(wù)時會將任務(wù)添加到該隊列中)中取出被取消的任務(wù),然后將其從格子的任務(wù)列表中移除牡借。
    3. 計算當(dāng)前tick所在的格子(bucket)
    4. 調(diào)用transferTimeoutsToBuckets方法將timeouts隊列中新建的任務(wù)轉(zhuǎn)移到所在格子的鏈表中
    5. 調(diào)用HashedWheelBucket.expireTimeouts方法執(zhí)行到期的任務(wù)

這里有幾個值的注意的數(shù)據(jù)結(jié)構(gòu):

  1. 任務(wù)并不是直接放在格子中的拳昌,而是維護(hù)了一個雙向鏈表,這種數(shù)據(jù)結(jié)構(gòu)非常便于插入和移除钠龙。
  2. 新添加的任務(wù)并不直接放入格子炬藤,而是先放入一個隊列中,這是為了避免多線程插入任務(wù)的沖突俊鱼。在每個tick運(yùn)行任務(wù)之前由worker線程自動對任務(wù)進(jìn)行歸集和分類刻像,插入到對應(yīng)的槽位里面。

https://blog.wangqi.love/articles/Java/%E5%AE%9A%E6%97%B6%E4%BB%BB%E5%8A%A1%E4%B8%8EHashedWheelTimer.html

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末并闲,一起剝皮案震驚了整個濱河市细睡,隨后出現(xiàn)的幾起案子,更是在濱河造成了極大的恐慌帝火,老刑警劉巖溜徙,帶你破解...
    沈念sama閱讀 217,509評論 6 504
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件,死亡現(xiàn)場離奇詭異犀填,居然都是意外死亡蠢壹,警方通過查閱死者的電腦和手機(jī),發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 92,806評論 3 394
  • 文/潘曉璐 我一進(jìn)店門九巡,熙熙樓的掌柜王于貴愁眉苦臉地迎上來图贸,“玉大人,你說我怎么就攤上這事冕广∈枞眨” “怎么了?”我有些...
    開封第一講書人閱讀 163,875評論 0 354
  • 文/不壞的土叔 我叫張陵撒汉,是天一觀的道長沟优。 經(jīng)常有香客問我,道長睬辐,這世上最難降的妖魔是什么挠阁? 我笑而不...
    開封第一講書人閱讀 58,441評論 1 293
  • 正文 為了忘掉前任,我火速辦了婚禮溯饵,結(jié)果婚禮上侵俗,老公的妹妹穿的比我還像新娘。我一直安慰自己丰刊,他們只是感情好坡慌,可當(dāng)我...
    茶點故事閱讀 67,488評論 6 392
  • 文/花漫 我一把揭開白布。 她就那樣靜靜地躺著藻三,像睡著了一般洪橘。 火紅的嫁衣襯著肌膚如雪跪者。 梳的紋絲不亂的頭發(fā)上,一...
    開封第一講書人閱讀 51,365評論 1 302
  • 那天熄求,我揣著相機(jī)與錄音渣玲,去河邊找鬼。 笑死弟晚,一個胖子當(dāng)著我的面吹牛忘衍,可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播卿城,決...
    沈念sama閱讀 40,190評論 3 418
  • 文/蒼蘭香墨 我猛地睜開眼枚钓,長吁一口氣:“原來是場噩夢啊……” “哼!你這毒婦竟也來了瑟押?” 一聲冷哼從身側(cè)響起搀捷,我...
    開封第一講書人閱讀 39,062評論 0 276
  • 序言:老撾萬榮一對情侶失蹤,失蹤者是張志新(化名)和其女友劉穎多望,沒想到半個月后嫩舟,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體,經(jīng)...
    沈念sama閱讀 45,500評論 1 314
  • 正文 獨居荒郊野嶺守林人離奇死亡怀偷,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點故事閱讀 37,706評論 3 335
  • 正文 我和宋清朗相戀三年家厌,在試婚紗的時候發(fā)現(xiàn)自己被綠了。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片椎工。...
    茶點故事閱讀 39,834評論 1 347
  • 序言:一個原本活蹦亂跳的男人離奇死亡饭于,死狀恐怖,靈堂內(nèi)的尸體忽然破棺而出维蒙,到底是詐尸還是另有隱情掰吕,我是刑警寧澤,帶...
    沈念sama閱讀 35,559評論 5 345
  • 正文 年R本政府宣布木西,位于F島的核電站畴栖,受9級特大地震影響随静,放射性物質(zhì)發(fā)生泄漏八千。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點故事閱讀 41,167評論 3 328
  • 文/蒙蒙 一燎猛、第九天 我趴在偏房一處隱蔽的房頂上張望恋捆。 院中可真熱鬧,春花似錦重绷、人聲如沸沸停。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,779評論 0 22
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽愤钾。三九已至瘟滨,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間能颁,已是汗流浹背杂瘸。 一陣腳步聲響...
    開封第一講書人閱讀 32,912評論 1 269
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機(jī)就差點兒被人妖公主榨干…… 1. 我叫王不留伙菊,地道東北人败玉。 一個月前我還...
    沈念sama閱讀 47,958評論 2 370
  • 正文 我出身青樓,卻偏偏與公主長得像镜硕,于是被迫代替她去往敵國和親运翼。 傳聞我的和親對象是個殘疾皇子,可洞房花燭夜當(dāng)晚...
    茶點故事閱讀 44,779評論 2 354

推薦閱讀更多精彩內(nèi)容