Netty源码分析之定时器HashedWheelTimer

徘徊笔记 2019-06-23

247

来源：https://github.com/netty/netty

这是一个环状的定时器，含有多个槽，任务就放到槽中，链表形式存储，标记时间是固定的，然后根据任务的具体定时时间与调度时间的关系分配到不同的槽中，达到定时执行的效果。实例化定时器，参数分别为线程工厂，标记时间，时间单位，槽数量，是否需要检查内存泄漏，最大等待执行的任务数。

public HashedWheelTimer(
        ThreadFactory threadFactory,
        long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
        long maxPendingTimeouts) {


    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);
    }


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


    // Convert tickDuration to nanos.
    this.tickDuration = unit.toNanos(tickDuration);


    // Prevent overflow.
    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));
    }
    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();
    }
}

创建槽数组，初始化为2的幂次方，这样可以使用位运算直接定位数组元素。

private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
    if (ticksPerWheel <= 0) {
        throw new IllegalArgumentException(
                "ticksPerWheel must be greater than 0: " + ticksPerWheel);
    }
    if (ticksPerWheel > 1073741824) {
        throw new IllegalArgumentException(
                "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
    }


    ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
    HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
    for (int i = 0; i < wheel.length; i ++) {
        wheel[i] = new HashedWheelBucket();
    }
    return wheel;
}


private static int normalizeTicksPerWheel(int ticksPerWheel) {
    int normalizedTicksPerWheel = 1;
    while (normalizedTicksPerWheel < ticksPerWheel) {
        normalizedTicksPerWheel <<= 1;
    }
    return normalizedTicksPerWheel;
}

创建并开始执行定时任务，设置任务的执行时间，判断任务的数量限制。

public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
    if (task == null) {
        throw new NullPointerException("task");
    }
    if (unit == null) {
        throw new NullPointerException("unit");
    }


    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;
    HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
    timeouts.add(timeout);
    return timeout;
}

启动定时线程，第一个任务会启动线程，第二个直接略过，然后等待线程启动后设置开始时间，用CountDownLatch等待，当开始时间不为0的话就代表启动初始化成功

private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
publicstatic final int WORKER_STATE_INIT = 0;
public static final int WORKER_STATE_STARTED = 1;
public static final int WORKER_STATE_SHUTDOWN = 2;           
public void start() {
    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.
    while (startTime == 0) {
        try {
            startTimeInitialized.await();
        } catch (InterruptedException ignore) {
            // Ignore - it will be ready very soon.
        }
    }
}

计算任务的时间与开始时间的间隔也就是相对截止时间，实例化为带有时间的超时任务类，保存到超时任务队列中。

private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue();
HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
    this.timer = timer;
    this.task = task;
    this.deadline = deadline;
}

定时线程执行任务，这里会获取当前服务器的启动时间差纳秒值，设置为线程的开始时间，然后CountDownLatch通知任务开始时间已经初始化完毕。

private final class Worker implements Runnable
private long tick;
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);
}

查看并等待下一个槽的任务截止时间到达即任务可以执行。标记tick初始值为0，计算当前下一个标记的截止时间，计算出需要sleep的时间差，当小于等于0时，说明时间已经到达，时间不足一毫秒且大于等于一纳秒的统一sleep一毫秒，如果当前时间溢出的话，任务就不能执行，当然long类型溢出的条件很苛刻，需要服务器运行Long.MAX_VALUE+1纳秒才可以，大约是292年。

private long waitForNextTick() {
    long deadline = tickDuration * (tick + 1);


    for (;;) {
        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;
            }
        }


        // Check if we run on windows, as if thats the case we will need
        // to round the sleepTime as workaround for a bug that only affect
        // the JVM if it runs on windows.
        //
        // See https://github.com/netty/netty/issues/356
        if (PlatformDependent.isWindows()) {
            sleepTimeMs = sleepTimeMs  10 * 10;
        }


        try {
            Thread.sleep(sleepTimeMs);
        } catch (InterruptedException ignored) {
            if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
                return Long.MIN_VALUE;
            }
        }
    }
}

当下一个标记的时间到达时返回当前相对时间，时间数为正数。使用位运算计算出当前需要执行的槽，执行已经取消的任务，也就是从取消任务队列中取出任务并移除

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);
            }
        }
    }
}

把任务放进对应的槽，每次最大出力100000个任务，避免有大量任务要转化而不能执行定时任务。从任务队列中取出任务，判断任务状态不能为取消状态，根据任务的截止时间与标记时间间隔计算出需要放置的标记位置，根据环状槽数量与当前标记位计算出该放置在那一圈的那个槽，这里有可能的位置已经被执行过了，也就是计算出来的标记位小于已经执行的标记位，所以这个时候就把该任务的标记位设置为当前标记位，然后通过位运算计算出该放到环数组的那个槽

private void transferTimeoutsToBuckets() {
    // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
    // adds new timeouts in a loop.
    for (int i = 0; i < 100000; i++) {
        HashedWheelTimeout timeout = timeouts.poll();
        if (timeout == null) {
            // all processed
            break;
        }
        if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
            // Was cancelled in the meantime.
            continue;
        }


        long calculated = timeout.deadline  tickDuration;
        timeout.remainingRounds = (calculated - tick)  wheel.length;


        final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
        int stopIndex = (int) (ticks & mask);


        HashedWheelBucket bucket = wheel[stopIndex];
        bucket.addTimeout(timeout);
    }
}

把该任务加到槽的链表中，这样就把分散的任务全部转化为具体槽的任务

private static final class HashedWheelBucket
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;
    }
}

执行该标记槽位的任务，判断当前槽是否有任务，依次判断链表节点的任务剩余轮次是否小于等于0，判断任务截止时间与当前的截止时间之间的大小关系，如果不满足剩余轮次的话就针对该任务的轮次减一操作。

public void expireTimeouts(long deadline) {
    HashedWheelTimeout timeout = head;


    // process all timeouts
    while (timeout != null) {
        HashedWheelTimeout next = timeout.next;
        if (timeout.remainingRounds <= 0) {
            next = remove(timeout);
            if (timeout.deadline <= deadline) {
                timeout.expire();
            } 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()) {
            next = remove(timeout);
        } else {
            timeout.remainingRounds --;
        }
        timeout = next;
    }
}

当该任务到达定时时间时，从该槽的链表中移除该任务，减少等待任务数量

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;
}

执行定时任务，最后标记位自增，准备执行下一个标记槽的任务

private static final class HashedWheelTimeout implements Timeout
private static final int ST_INIT = 0;
private static final int ST_CANCELLED = 1;
private static final int ST_EXPIRED = 2;
private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
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);
        }
    }
}

数据库

文章转载自徘徊笔记，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

Netty源码分析之定时器HashedWheelTimer

评论