一、案例分析
public void testExecutionOnTaskExecutor() throws InterruptedException {
int timeout = 10;
final CountDownLatch latch = new CountDownLatch(1);
final CountDownLatch timeoutLatch = new CountDownLatch(1);
Executor executor = new Executor() {
@Override
public void execute(Runnable command) {
try {
command.run();
} finally {
latch.countDown();
}
}
};
final HashedWheelTimer timer = new HashedWheelTimer(Executors.defaultThreadFactory(), 100,
TimeUnit.MILLISECONDS, 32, true, 2, executor);
timer.newTimeout(new TimerTask() {
@Override
public void run(final Timeout timeout) throws Exception {
timeoutLatch.countDown();
}
}, timeout, TimeUnit.MILLISECONDS);
latch.await();
timeoutLatch.await();
timer.stop();
}二、底层接口
public interface Timer {
/**
* Schedules the specified {@link TimerTask} for one-time execution after
* the specified delay.
*
* @return a handle which is associated with the specified task
*
* @throws IllegalStateException if this timer has been {@linkplain #stop() stopped} already
* @throws RejectedExecutionException if the pending timeouts are too many and creating new timeout
* can cause instability in the system.
*/
Timeout newTimeout(TimerTask task, long delay, TimeUnit unit);
/**
* Releases all resources acquired by this {@link Timer} and cancels all
* tasks which were scheduled but not executed yet.
*
* @return the handles associated with the tasks which were canceled by
* this method
*/
Set<Timeout> stop();
}
newTimeOut: 提交任务，并在延迟delay的unit单位时间后，执行任务
stop:取消所有调度中但是未执行的任务
三、构建hashWheel函数
public HashedWheelTimer(
ThreadFactory threadFactory,
long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
long maxPendingTimeouts, Executor taskExecutor)threadFactory：线程工厂，用于创建工作线程
tickDuration：每次转动的间隔的时间
unit：每次转动的间隔的时间的时间单元
ticksPerWheel：时间轮上一共有多少个 slot，默认 512 个。分配的 slot 越多，占用的内存空间就越大
leakDetection：是否开启内存泄漏检测
maxPendingTimeouts：最大等待任务数
taskExecutor:任务处理器
3.1 HashedWheelBucket[] wheel
// 时间轮的环形数组
wheel = createWheel(ticksPerWheel);
private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
//ticksPerWheel may not be greater than 2^30
checkInRange(ticksPerWheel, 1, 1073741824, "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 final class HashedWheelBucket {
// Used for the linked-list datastructure
private HashedWheelTimeout head;
private HashedWheelTimeout tail;
。。。。
}通过构造方法可以得知，其实这个类似于HashMap底层是数组+链表的结构，构造出了一个2^n的数组，位数计算是效率最高的算法
一直左移1位，即相当于*2,即该代码是为了得出 >= ticksPerWheel的2^n值，如ticksPerWheel
为5，则结果值为8
private static int normalizeTicksPerWheel(int ticksPerWheel) {
int normalizedTicksPerWheel = 1;
while (normalizedTicksPerWheel < ticksPerWheel) {
normalizedTicksPerWheel <<= 1;
}
return normalizedTicksPerWheel;
}
得出2^n次大小的数组后，即可通过[(2^n)-1] & index 进行求模，所以记录一个值
mask = wheel.length - 1; 对应结构如下：
3.2 tickDuration
// Convert tickDuration to nanos.
// 转化为纳秒
long duration = unit.toNanos(tickDuration);
// Prevent overflow.
if (duration >= Long.MAX_VALUE / wheel.length) {
throw new IllegalArgumentException(String.format(
"tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
tickDuration, Long.MAX_VALUE / wheel.length));
}
if (duration < MILLISECOND_NANOS) {
logger.warn("Configured tickDuration {} smaller than {}, using 1ms.",
tickDuration, MILLISECOND_NANOS);
this.tickDuration = MILLISECOND_NANOS;
} else {
this.tickDuration = duration;
}根据long tickDuration, TimeUnit unit 得出每次hash时钟摆动的间隔值
3.3 workerThread
workerThread = threadFactory.newThread(worker); 根据线程工厂，构建出工作线程，也就是通过该线程进行了延迟任务的执行
3.4 other
leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
// 最大允许等待任务数，HashedWheelTimer 中任务超出该阈值时会抛出异常
this.maxPendingTimeouts = maxPendingTimeouts;
// 如果实例数超过 64，会打印错误日志
if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
reportTooManyInstances();
}
private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger(); 类常量
最多创建64个实例？
HashedWheelTimer is a shared resource that must be reused across the JVM
so that only a few instances are created
根据报错提示，可以知道这是一个共享资源，可以进行reused，所以只需要创建少量就行3.5 疑问
通过初始化函数得知，只是创建了工作线程，并未进行工作线程的执行，那么工作线程的start在什么时候调用？以及内存泄漏检测原理？
4 添加任务
@Override
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
checkNotNull(task, "task");
checkNotNull(unit, "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;
// Guard against overflow.
if (delay > 0 && deadline < 0) {
deadline = Long.MAX_VALUE;
}
HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);//创建定时任务
timeouts.add(timeout); // 加入队列
return timeout;
}private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue();// Mpsc Queue 可以理解为多生产者单消费者的线程安全队列后续介绍
小结：newTimeout() 方法主要做了三件事，分别为启动工作线程，创建定时任务，并把任务添加到 Mpsc Queue。HashedWheelTimer 的工作线程采用了懒启动的方式，不需要用户显示调用。这样做的好处是在时间轮中没有任务时，可以避免工作线程空转而造成性能损耗。
// Initialize the startTime. startTime = System.nanoTime(); startTime 是在任务开始执行后，才进行赋值的所以要添加阻塞startTimeInitialized 进行阻塞。
4.1 start()
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.
}
}
}
4.2 疑问
什么时候进行时间轮的执行？workThread执行
5 工作线程的执行
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(); // 计算下次 tick 的时间, 然后sleep 到下次 tick
if (deadline > 0) { // 可能因为溢出或者线程中断，造成 deadline <= 0
int idx = (int) (tick & mask); // 获取当前 tick 在 HashedWheelBucket 数组中对应的下标
processCancelledTasks(); // 移除被取消的任务
HashedWheelBucket bucket =
wheel[idx];
transferTimeoutsToBuckets();//从 Mpsc Queue 中取出任务加入对应的 slot 中
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.
// 时间轮退出后，取出 slot 中未执行且未被取消的任务，并加入未处理任务列表，以便 stop() 方法返回
for (HashedWheelBucket bucket: wheel) {
bucket.clearTimeouts(unprocessedTimeouts);
}
// 将还没添加到 slot 中的任务取出，如果任务未取消则加入未处理任务列表，以便 stop() 方法返回
for (;;) {
HashedWheelTimeout timeout = timeouts.poll();
if (timeout == null) {
break;
}
if (!timeout.isCancelled()) {
unprocessedTimeouts.add(timeout);
}
}
processCancelledTasks();
}前面提到start()方法中会启动worker线程，并且会等待startTime不为0，worker线程会把startTime设置为当前的纳秒时间，并且startTimeInitialized.countDown()唤醒阻塞在start()方法的线程。
5.1 waitForNextTick
private long waitForNextTick() {
long deadline = tickDuration * (tick + 1); // (根据当前tick+1) * tickDuration 每次转动的间隔的时间得出截至时间，如当前ticket为1，间隔为1s，则deadline 为2*1 = 2s，表示2s后执行
for (;;) {
final long currentTime = System.nanoTime() - startTime; // 当前时间-开始时间表示当前已经过了currentTime 的时间差值，deadline - currentTime则可以得出还剩多少时间可以进行等待，因为转为了纳秒为单位，所以只要还有1纳秒，就需要进行继续sleep，所以进行+ 999999的流程
long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
// 若小于0，则表示需要立刻执行，返回currentTime差值时间
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;
if (sleepTimeMs == 0) {
sleepTimeMs = 1;
}
}
try {
Thread.sleep(sleepTimeMs);
} catch (InterruptedException ignored) {
if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
return Long.MIN_VALUE;
}
}
}
}5.1 transferTimeoutsToBuckets
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);
}
}
transferTimeoutsToBuckets() 的主要工作就是从 Mpsc Queue 中取出任务，然后添加到时间轮对应的 HashedWheelBucket 中。每次时针 tick 最多只处理 100000 个任务，一方面避免取任务的操作耗时过长，另一方面为了防止执行太多任务造成 Worker 线程阻塞。
根据用户设置的任务 deadline，可以计算出任务需要经过多少次 tick 才能开始执行 以及需要在时间轮中转动圈数 remainingRounds，remainingRounds 会记录在 HashedWheelTimeout 中，在执行任务的时候 remainingRounds进行判断当前是否执行。因为时间轮中的任务并不能够保证及时执行，假如有一个任务执行的时间特别长，那么任务在 timeouts 队列里已经过了执行时间，也没有关系，Worker 会将这些任务直接加入当前HashedWheelBucket 中，所以过期的任务并不会被遗漏。
5.2 HashedWheelBucket#expireTimeouts()
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 --; // 未到执行时间，remainingRounds 减 1
}
timeout = next;
}
}
执行任务即： taskExecutor进行
execute方法
public void expire() {
if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
return;
}
try {
timer.taskExecutor.execute(this);
} catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("An exception was thrown while submit " + TimerTask.class.getSimpleName()
+ " for execution.", t);
}
}
}- HashedWheelTimeout，任务的封装类，包含任务的到期时间 deadline、需要经历的圈数 remainingRounds 等属性。
- HashedWheelBucket，相当于时间轮的每个 slot，内部采用双向链表保存了当前需要执行的 HashedWheelTimeout 列表。
- Worker，HashedWheelTimer 的核心工作引擎，负责处理定时任务。
6 缺点
- 如果长时间没有到期任务，那么会存在时间轮空推进的现象。（通过固定的时间间隔 tickDuration ）
- 只适用于处理耗时较短的任务，由于 Worker 是单线程的，如果一个任务执行的时间过长，会造成 Worker 线程阻塞。
- 相比传统定时器的实现方式，内存占用较大。
