Netty 时间轮算法实现HashedWheelTimer Netty

1.HashedWheelTimer 参数解析

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

1.ThreadFactory 自定义线程工厂，用于创建线程对象。 2.tickDuration间隔多久走到下一槽（相当于时钟走一格） 3.unit 定义tickDuration的时间单位 4.ticksPerWheel 一圈有多个槽 5.leakDetection 是否开启内存泄漏检测。 6. maxPendingTimeouts 最多待执行的任务个数。0或负数表示无限制。
1.创建的槽位个数必须是2的次方，比如我传入的ticksPerWheel 是60，最后实际是64。

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

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

2.tickDuration 转化为纳秒

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

转一圈的值不能大于Long.MAX_VALUE

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

3.HashedWheelTimer的实例不能超过64

private static final int INSTANCE_COUNT_LIMIT = 64;

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

4.添加新任务maxPendingTimeouts 的检查限制

public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit)

中有如下代码

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

long deadline = System.nanoTime() + unit.toNanos(delay) - startTime; HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline); timeouts.add(timeout); return timeout;

private final Queue timeouts = PlatformDependent.newMpscQueue();

最后封装成HashedWheelTimout添加到队列（timeouts）中 2.工作线程

workerThread = threadFactory.newThread(worker);

private final Worker worker = new Worker();

我们看Worker关键代码

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

我们看下面这个方法，它将队列（newTask的时候添加任务的那个队列）中的任务放到对应的HashedWheelBucket中。

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

计算需要轮询多少轮（remainingRounds），以及哪个槽（stopIndex），放入对应的HashedWheelBucked中。
执行当前槽的到期任务

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

如果是到期了，则执行任务，否则将remainingRounds 减一。执行任务的方法：

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

3.总结

文章图片

当添加新的延迟任务时（newTimeOut），将任务封装成HashedWheelTimeout对象，然后添加到timeouts队列中，工作线程（Worker）每间隔tickDuration时间向前走一格，找到当前槽位中对应的对象（HashedWheelBucket链表），然后将timeouts队列中的前100000个任务放入对应的槽位中，最后遍历当前槽位中任务看是否过期可以执行了。

HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(new ThreadFactory() { @Override public Thread newThread(Runnable r) { return new Thread(r, "HashedWheelTimerTest"); } }, 1, TimeUnit.SECONDS, 60, true, 10); hashedWheelTimer.start(); final long millis = System.currentTimeMillis(); hashedWheelTimer.newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { System.out.println("done:" + (System.currentTimeMillis() - millis)); System.out.println(timeout); } }, 1, TimeUnit.SECONDS);