背景
最近有接觸netty相關(guān)內(nèi)容，組內(nèi)有做關(guān)于netty時(shí)間輪的分享招狸，正好總結(jié)這篇文章，做個(gè)了解和記錄。時(shí)間輪在超時(shí)控制进鸠，異常處理司恳，鎖控制等方面都有非常多的應(yīng)用猎塞。本期要說(shuō)的netty時(shí)間輪是實(shí)現(xiàn)相對(duì)簡(jiǎn)單的一種，比較復(fù)雜的如kafka是多級(jí)的時(shí)間輪實(shí)現(xiàn)杠纵，我們暫不做介紹荠耽。
原理
首先我們理解什么是時(shí)間輪，其實(shí)可以形象的看做是一個(gè)手表比藻，按照固定的時(shí)間铝量，即tick，每次擺動(dòng)對(duì)應(yīng)一個(gè)階段的時(shí)間银亲，然后對(duì)應(yīng)時(shí)間刻度上綁定待觸發(fā)任務(wù)慢叨。當(dāng)時(shí)鐘指針到達(dá)時(shí)，對(duì)應(yīng)的任務(wù)就可以執(zhí)行了群凶。比如一種場(chǎng)景插爹，每一個(gè)刻度代表一秒，某個(gè)超時(shí)判斷為10s，則這個(gè)超時(shí)就在10s的刻度上（假定時(shí)鐘一個(gè)周期大于10s赠尾，即格子數(shù)大于10個(gè)）力穗。當(dāng)指針從0到達(dá)10s的刻度，就是該超時(shí)判斷執(zhí)行的時(shí)候气嫁。
時(shí)間輪
以上是個(gè)簡(jiǎn)單時(shí)間輪的原理当窗，實(shí)用的時(shí)間輪比之要復(fù)雜。
思路參考
關(guān)于時(shí)間輪的實(shí)現(xiàn)寸宵，基本都基于George Varghese和Tony Lauck的論文崖面，后面有鏈接。netty的時(shí)間輪處理除了上面的原理梯影，還在每個(gè)調(diào)度任務(wù)中記錄輪次巫员，在輪次減少到0后，再判斷當(dāng)前輪上的剩余格數(shù)甲棍，進(jìn)行任務(wù)執(zhí)行简识。
注意使用的netty版本是
4.1.51.Final
netty實(shí)現(xiàn)代碼
- 時(shí)間輪構(gòu)造
public HashedWheelTimer(
ThreadFactory threadFactory,
long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection) {
this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection, -1);
}
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.
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) {
if (logger.isWarnEnabled()) {
logger.warn("Configured tickDuration %d smaller then %d, using 1ms.",
tickDuration, MILLISECOND_NANOS);
}
this.tickDuration = MILLISECOND_NANOS;
} else {
this.tickDuration = duration;
}
//創(chuàng)建調(diào)度線程
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();
}
}
除了一些參數(shù)校驗(yàn)，我們按照順序感猛，先看主要的方法createWheel七扰，按照時(shí)間輪每圈的格子數(shù)，先做個(gè)轉(zhuǎn)換normalizeTicksPerWheel()陪白，找到最近的一個(gè)二進(jìn)制數(shù)值颈走，如7取8,13取16。這么做主要是通過(guò)利用&運(yùn)算可以快速取余咱士，用到了后面的定義的mask= wheel.length-1立由。這種方式在HashMap以及一些中間件中經(jīng)常使用。隨后司致，根據(jù)新的每圈格子數(shù)拆吆，創(chuàng)建對(duì)應(yīng)的拉鏈數(shù)組。繼續(xù)看構(gòu)造函數(shù)脂矫，會(huì)創(chuàng)建一個(gè)worker線程，用來(lái)遍歷時(shí)間輪霉晕，執(zhí)行調(diào)度庭再。
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;
}
- worker線程
worker線程我們主要看下run方法。獲得當(dāng)前tick的一個(gè)過(guò)期時(shí)間牺堰，然后取余得到當(dāng)前tick的分桶拄轻，然后，指定的bucket再判斷當(dāng)前tick的截止時(shí)間伟葫，判斷輪次是否截止+是否取消恨搓，是否得處于過(guò)期timeout.deadline <= deadline。具體可以看io.netty.util.HashedWheelTimer.HashedWheelBucket.expireTimeouts
private final class Worker implements Runnable {
private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
private long tick;
@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();
}
- 時(shí)間輪每個(gè)格子對(duì)應(yīng)的分桶
如下，對(duì)應(yīng)到時(shí)間輪每個(gè)格子斧抱，其中的元素是實(shí)際調(diào)度任務(wù)常拓，構(gòu)造了一個(gè)雙向鏈表的結(jié)構(gòu)。其中expireTimeouts辉浦，即上面介紹的當(dāng)指定的tick到來(lái)時(shí)弄抬，判斷并執(zhí)行分桶上的調(diào)度任務(wù)。
private static final class HashedWheelBucket {
// Used for the linked-list datastructure
private HashedWheelTimeout head;
private HashedWheelTimeout tail;
/**
* Add {@link HashedWheelTimeout} to this bucket.
*/
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;
}
}
/**
* Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
*/
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;
}
}
- 實(shí)際調(diào)度任務(wù)元素
下面是實(shí)際執(zhí)行的調(diào)度任務(wù)元素宪郊，它實(shí)現(xiàn)了javaTimeout接口掂恕，元素包括對(duì)時(shí)間輪的索引，調(diào)度任務(wù)弛槐，本身的截止時(shí)間懊亡，以及狀態(tài)，輪次乎串，分桶和前后鏈表指針等店枣。除了cancel,remove等方法，核心的是expire()用來(lái)真正實(shí)現(xiàn)任務(wù)的調(diào)度灌闺。每個(gè)有自己對(duì)應(yīng)的deadline艰争，即預(yù)計(jì)調(diào)度的時(shí)間點(diǎn)。
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");
private final HashedWheelTimer timer;
private final TimerTask task;
private final long deadline;
@SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
private volatile int state = ST_INIT;
// remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
// HashedWheelTimeout will be added to the correct HashedWheelBucket.
long remainingRounds;
// This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
// As only the workerThread will act on it there is no need for synchronization / volatile.
HashedWheelTimeout next;
HashedWheelTimeout prev;
// The bucket to which the timeout was added
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() {
// only update the state it will be removed from HashedWheelBucket on next tick.
if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
return false;
}
// If a task should be canceled we put this to another queue which will be processed on each tick.
// So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way
// we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible.
timer.cancelledTimeouts.add(this);
return true;
}
void remove() {
HashedWheelBucket bucket = this.bucket;
if (bucket != null) {
bucket.remove(this);
} else {
timer.pendingTimeouts.decrementAndGet();
}
}
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);
}
}
}
- 添加調(diào)度任務(wù)
這里添加任務(wù)就是創(chuàng)建實(shí)際調(diào)度的對(duì)象桂对，時(shí)間輪的啟動(dòng)也在這里start()甩卓，最后創(chuàng)建調(diào)度元素后，添加到一個(gè)隊(duì)列里蕉斜，然后在調(diào)度的時(shí)候逾柿，再?gòu)年?duì)列轉(zhuǎn)換到buckets中。這個(gè)隊(duì)列是使用timeouts = PlatformDependent.newMpscQueue();jclTools創(chuàng)建的宅此。是一個(gè)高性能的并發(fā)Queue包机错。
@Override
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;
// Guard against overflow.
if (delay > 0 && deadline < 0) {
deadline = Long.MAX_VALUE;
}
HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
timeouts.add(timeout);
return timeout;
}
- 啟動(dòng)
時(shí)間輪對(duì)象本身的start()方法是public的，但是在添加調(diào)度任務(wù)的時(shí)候啟動(dòng)即可父腕，不用顯式地調(diào)用弱匪，畢竟如果時(shí)間輪沒(méi)有任務(wù)，也沒(méi)有啟動(dòng)的必要璧亮。注意這里判斷狀態(tài)都是原子類萧诫，其中還主要用到了AtomicIntegerFieldUpdater，是jdk基于反射實(shí)現(xiàn)的原子狀態(tài)管理類枝嘶。
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.
}
}
}
執(zhí)行示例
import io.netty.util.HashedWheelTimer;
import org.junit.Test;
import java.time.LocalDateTime;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Test
public void testTimeWheel() throws InterruptedException {
HashedWheelTimer wheelTimer = new HashedWheelTimer(Executors.defaultThreadFactory(), 1, TimeUnit.SECONDS, 64, false);
System.out.println("time wheel start@" + LocalDateTime.now());
wheelTimer.newTimeout(timeout -> System.out.println("timeout 10s --> " + LocalDateTime.now()), 10, TimeUnit.SECONDS);
wheelTimer.newTimeout(timeout -> System.out.println("timeout 20s --> " + LocalDateTime.now()), 20, TimeUnit.SECONDS);
Thread.currentThread().join();
}
運(yùn)行結(jié)果:
time wheel start@2020-11-13T18:10:03.232
timeout 10s --> 2020-11-13T18:10:14.234
timeout 20s --> 2020-11-13T18:10:24.233
總結(jié)
以上就是本期的全部?jī)?nèi)容帘饶，對(duì)netty時(shí)間輪算是有了個(gè)初步的了解。對(duì)于其中一些更加深入的細(xì)節(jié)群扶，還需要再努力研究下及刻。感謝閱讀镀裤。以下參考資料感興趣的讀者可以多做深入。
