Netty 工具類 —— HashedWheelTimer 講解

一、前言

在網絡通信中管理上萬的連接，每個連接都有超時任務，如果為每個任務啟動一個TImer超時器，那么會占用大量資源。為了解決這個問題，可用Netty工具類HashedWheelTimer。

 

二、HashedWheelTimer原理

1.概論

（學習一個類，最好的方式是看api文檔或源碼的注釋，我下載了Netty源碼）

這個類用來計划執行非精准的I/O超時。可以通過指定每一格的時間間隔來改變執行時間的精確度。在大多數網絡應用中，I/O超時不需要十分准確，因此，默認的時間間隔是100 毫秒，這個值適用於大多數場合。HashedWheelTimer內部結構可以看做是個車輪，簡單來說，就是TimerTask的hashTable的車輪。車輪的size默認是512，可以通過構造函數自己設置這個值。注意，當HashedWheelTimer被實例化啟動后，會創建一個新的線程，因此，你的項目里應該只創建它的唯一一個實例。

（這個類的源自一位教授的論文 Tony Lauck's paper。算法是代碼的靈魂，最難的是算法）

 

2.結構

下方圖示闡述了大致的結構模型

Demo

@Test(timeout = 50000000)
    public void testTimerOverflowWheelLength() throws InterruptedException {
        final HashedWheelTimer timer = new HashedWheelTimer(
            Executors.defaultThreadFactory(), 100, TimeUnit.MILLISECONDS, 32);

        timer.newTimeout(new TimerTask() {
            @Override
            public void run(final Timeout timeout) throws Exception {
                System.out.println("lee");   //打印名字
            }
        }, 1000, TimeUnit.MILLISECONDS);

        Thread.sleep(10000);
        assertFalse(timer.stop().isEmpty());
    }

通過上面demo，我再說明一下hashedWheelTimer類的構造。

超時任務1000毫秒，超時之后，由hashedWheelTimer類中的worker線程，執行超時之后的任務（打印名字）。hashedWheelTimer有32個槽（相當於一個圓的32分之一），每移動一個槽的時間是100毫秒。
任務需要經過的tick數為: 1000 / 100 = 10次         (等待時長 / tickDuration)
任務需要經過的輪數為  : 10次 / 32次/輪 = 0輪     (tick總次數 / ticksPerWheel)
因為任務超時后不能馬上被worker線程執行，需要等到worker線程移到相應卡槽位置時才會執行，因此說執行時間不精確。

hashedWheelTimer的核心是Worker線程，主要負責每過tickDuration時間就累加一次tick. 同時, 也負責執行到期的timeout任務, 此外，還負責添加timeou任務到指定的wheel中。

接下看看源碼部分。

 

構造器

構造器的各個參數的說明和注釋十分詳細，應該沒有不好理解的地方。

    /**
     * Creates a new timer.
     *
     * @param threadFactory        a {@link ThreadFactory} that creates a
     *                             background {@link Thread} which is dedicated to    用來創建后台線程
     *                             {@link TimerTask} execution.
     * @param tickDuration         the duration between tick                          每格時間間隔 默認 100
     * @param unit                 the time unit of the {@code tickDuration}          時間單位 默認 毫秒
     * @param ticksPerWheel        the size of the wheel                              輪子size（一圈多少格）
                                                                                         默認 512 如果不是2的N次方，則去大於該參數的第一個2的N次方
                                                                                         理由 便於Hash值的計算10      * @param leakDetection        {@code true} if leak detection should be enabled always,
     *                             if false it will only be enabled if the worker thread is not
     *                             a daemon thread.
     * @param  maxPendingTimeouts  The maximum number of pending timeouts after which call to
     *                             {@code newTimeout} will result in
     *                             {@link java.util.concurrent.RejectedExecutionException}
     *                             being thrown. No maximum pending timeouts limit is assumed if
     *                             this value is 0 or negative.                       最大待定超時時間 默認 不設置
     * @throws NullPointerException     if either of {@code threadFactory} and {@code unit} is {@code null}
     * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
     */
    public HashedWheelTimer(
            ThreadFactory threadFactory,
            long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
            long maxPendingTimeouts) {
           ...

        // Normalize ticksPerWheel to power of two and initialize the wheel.
        wheel = createWheel(ticksPerWheel);
        mask = wheel.length - 1;

        // Convert tickDuration to nanos.
        long duration = unit.toNanos(tickDuration);
           ...52 
        if (duration < MILLISECOND_NANOS) {
            if (logger.isWarnEnabled()) {
                logger.warn("Configured tickDuration %d smaller then %d, using 1ms.",
                            tickDuration, MILLISECOND_NANOS);
            }
            this.tickDuration = MILLISECOND_NANOS;
        } else {
            this.tickDuration = duration;
        }
        // 這里創建worker線程，worker線程是重點。
        workerThread = threadFactory.newThread(worker);

        leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;

        this.maxPendingTimeouts = maxPendingTimeouts;

        if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
            WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
            reportTooManyInstances();
        }
    }

我們接下來看newTimeout（）方法

    @Override
    public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
           ... 9 
        long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();

        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.
        long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;

        // Guard against overflow.
        if (delay > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }
        HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
        timeouts.add(timeout);
        return timeout;
    }

我們接着進到worker線程的啟動：

/**
     * Starts the background thread explicitly.  The background thread will
     * start automatically on demand even if you did not call this method.
     *
     * @throws IllegalStateException if this timer has been
     *                               {@linkplain #stop() stopped} already
     */
    public void start() {     // 這個方法為什么是public的？我們可以在實例化HashedWheelTimer后，主動調用這個方法，啟動worker線程  9         switch (WORKER_STATE_UPDATER.get(this)) {
            case WORKER_STATE_INIT:
                if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
 workerThread.start();            // 在這里啟動了worker線程 13                 }
                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.
           // 為了等待worker線程初始化startTime
        while (startTime == 0) {
            try {
                startTimeInitialized.await();
            } catch (InterruptedException ignore) {
                // Ignore - it will be ready very soon.
            }
        }
    }

 下一步，我們看看worker線程里面的操作。

首先是初始化startTime，CountDownLatch的觸發，后面do while操作可以看作是圓盤在一個個轉動，每轉一個就會用worker線程處理，格子中超時的任務。

        @Override
        public void run() {
            // Initialize the startTime.
            startTime = System.nanoTime();
            if (startTime == 0) {
                // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
                startTime = 1;
            }

            // Notify the other threads waiting for the initialization at start().
            startTimeInitialized.countDown();

            do {
                final long deadline = waitForNextTick();
                if (deadline > 0) {
                    int idx = (int) (tick & mask);
                    processCancelledTasks();
                    HashedWheelBucket bucket =
                            wheel[idx];
                    transferTimeoutsToBuckets();
                    bucket.expireTimeouts(deadline);
                    tick++;
                }
            } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);

            // Fill the unprocessedTimeouts so we can return them from stop() method.
            for (HashedWheelBucket bucket: wheel) {
                bucket.clearTimeouts(unprocessedTimeouts);
            }
            for (;;) {
                HashedWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    break;
                }
                if (!timeout.isCancelled()) {
                    unprocessedTimeouts.add(timeout);
                }
            }
            processCancelledTasks();
        }

 

 

----------------------------------------------------------------------------------------------

 參考資源：

https://www.jianshu.com/p/328f22432638

https://my.oschina.net/haogrgr/blog/489320

https://chuansongme.com/n/1650380646616