前言
好久沒寫文章了院溺，最近沒事兒看了下Redisson里面的分布式鎖的寫法眉菱，進(jìn)而看到了它使用了netty中的HashedWheelTimer，大致掃了一下拐格，覺得有點(diǎn)意思僧免，花了點(diǎn)時(shí)間看了下代碼，把自己的一些感想寫出來(lái)捏浊，供大家參考一下懂衩。
一圖勝千言
netty中的HashedWheelTimer基于這篇論文，首先我們確定金踪，HashedWheelTimer提供的是一個(gè)定時(shí)任務(wù)的一個(gè)優(yōu)化實(shí)現(xiàn)方案浊洞，在netty中主要用于異步IO的定時(shí)規(guī)劃觸發(fā)（A timer optimized for approximated I/O timeout scheduling）。為了方便大家理解胡岔，可以先來(lái)看看我畫的這個(gè)圖
HashedWheelTimer
這個(gè)圖基本上就涵蓋了HashedWheelTimer的所有的概念要素：
- workerThread 單線程用于處理所有的定時(shí)任務(wù)法希，它會(huì)在每個(gè)tick執(zhí)行一個(gè)bucket中所有的定時(shí)任務(wù)，以及一些其他的操作靶瘸。意味著定時(shí)任務(wù)不能有較大的阻塞和耗時(shí)苫亦，不然就會(huì)影響定時(shí)任務(wù)執(zhí)行的準(zhǔn)時(shí)性和有效性。
-
wheel 一個(gè)時(shí)間輪怨咪，其實(shí)就是一個(gè)環(huán)形數(shù)組屋剑，數(shù)組中的每個(gè)元素代表的就是未來(lái)的某些時(shí)間片段上需要執(zhí)行的定時(shí)任務(wù)的集合。
這里需要注意的就是不是某個(gè)時(shí)間而是某些時(shí)間惊暴。因?yàn)楸确秸f(shuō)我時(shí)間輪上的大小是10饼丘，時(shí)間間隔是1s，那么我1s和11s的要執(zhí)行的定時(shí)任務(wù)都會(huì)在index為1的格子上辽话。 - tick 工作線程當(dāng)前運(yùn)行的tick數(shù)肄鸽，每一個(gè)tick代表worker線程當(dāng)前的一次工作時(shí)間
- hash 在時(shí)間輪上的hash函數(shù)。默認(rèn)是tick%bucket的數(shù)量油啤，即將某個(gè)時(shí)間節(jié)點(diǎn)映射到了時(shí)間輪上的某個(gè)唯一的格子上典徘。
- bucket 時(shí)間輪上的一個(gè)格子，它維護(hù)的是一個(gè)Timeout的雙向鏈表益咬，保存的是這個(gè)哈希到這個(gè)格子上的所有Timeout任務(wù)逮诲。
- timeout 代表一個(gè)定時(shí)任務(wù)，其中記錄了自己的deadline，運(yùn)行邏輯以及在bucket中需要呆滿的圈數(shù)梅鹦，比方說(shuō)之前1s和11s的例子裆甩，他們對(duì)應(yīng)的timeout中圈數(shù)就應(yīng)該是0和1。 這樣當(dāng)遍歷一個(gè)bucket中所有的timeout的時(shí)候齐唆，只要圈數(shù)為0說(shuō)明就應(yīng)該被執(zhí)行嗤栓，而其他情況就把圈數(shù)-1就好。
除此之外箍邮，netty的HashedWheelTimer實(shí)現(xiàn)還有兩個(gè)東西值得關(guān)注茉帅，分別是pending-timeouts隊(duì)列和cancelled-timeouts隊(duì)列。這兩個(gè)隊(duì)列分別記錄新添加的定時(shí)任務(wù)和要取消的定時(shí)任務(wù)锭弊，當(dāng)workerThread每次循環(huán)運(yùn)行時(shí)堪澎，它會(huì)先將pending-timeouts隊(duì)列中一定數(shù)量的任務(wù)移動(dòng)到它們對(duì)應(yīng)的bucket，并取消掉cancelled-timeouts中所有的任務(wù)味滞。由于添加和取消任務(wù)可以由任意線程發(fā)起樱蛤，而相應(yīng)的處理只會(huì)在workerThread里，所以為了進(jìn)一步提高性能桃犬，這兩個(gè)隊(duì)列都是用了JCTools里面的MPSC(multiple-producer-single-consumer)隊(duì)列刹悴。
擼碼
看完了圖，咱們就來(lái)擼碼了攒暇。在擼碼之前土匀，咱們?cè)賮?lái)看個(gè)圖
HashedWheelTimer UML
這個(gè)圖是為我根據(jù)netty HashedWheelTimer代碼結(jié)構(gòu)整理的一個(gè)UML結(jié)構(gòu)圖，大家看代碼的時(shí)候可以以這個(gè)圖作為一個(gè)參考借鑒形用。比較簡(jiǎn)單的父接口咱們就不去了就轧，只需要知道幾點(diǎn)：
- TimerTask是一個(gè)定時(shí)任務(wù)的實(shí)現(xiàn)接口，其中run方法包裝了定時(shí)任務(wù)的邏輯
- Timeout是一個(gè)定時(shí)任務(wù)提交到Timer之后返回的句柄田度，通過這個(gè)句柄外部可以取消這個(gè)定時(shí)任務(wù)妒御，并對(duì)定時(shí)任務(wù)的狀態(tài)進(jìn)行一些基本的判斷
- Timer是HashedWheelTimer實(shí)現(xiàn)的父接口，僅定義了如何提交定時(shí)任務(wù)和如何停止整個(gè)定時(shí)機(jī)制
接下來(lái)咱們就去HashedWheelTimer去走一遭
HashedWheelTimer
我們首先來(lái)看看構(gòu)造函數(shù)（下文中所有注釋均為本人書寫）
public HashedWheelTimer(
ThreadFactory threadFactory,
long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
long maxPendingTimeouts) {
// 線程工廠镇饺，用于創(chuàng)建我們的worker線程
if (threadFactory == null) {
throw new NullPointerException("threadFactory");
}
// 一個(gè)tick的時(shí)間單位
if (unit == null) {
throw new NullPointerException("unit");
}
// 一個(gè)tick的時(shí)間間隔
if (tickDuration <= 0) {
throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
}
// 時(shí)間輪上一輪有多少個(gè)tick/bucket
if (ticksPerWheel <= 0) {
throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
}
// 將時(shí)間輪的大小規(guī)范化到2的n次方乎莉，這樣可以用位運(yùn)算來(lái)處理mod操作，提高效率
wheel = createWheel(ticksPerWheel);
// 計(jì)算位運(yùn)算需要的掩碼
mask = wheel.length - 1;
// 轉(zhuǎn)換時(shí)間間隔到納秒
long duration = unit.toNanos(tickDuration);
// 防止溢出
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));
}
// 時(shí)間間隔至少要1ms
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)建worker線程
workerThread = threadFactory.newThread(worker);
// 處理泄露監(jiān)控（本篇文章不涉及）
leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
// 設(shè)置最大等待任務(wù)數(shù)
this.maxPendingTimeouts = maxPendingTimeouts;
// 限制timer的實(shí)例數(shù)奸笤，避免過多的timer線程反而影響性能
if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
reportTooManyInstances();
}
}
構(gòu)造函數(shù)的調(diào)理比較清晰惋啃，這其中我們需要關(guān)注的可能就只有createWheel方法：
private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
// 處理時(shí)間輪太小或者太大造成的異常
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);
}
// 規(guī)范化到2的n次方
ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
// 創(chuàng)建每個(gè)bucket
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;
// 不斷地左移位直到找到大于等于時(shí)間輪大小的2的n次方出現(xiàn)
while (normalizedTicksPerWheel < ticksPerWheel) {
normalizedTicksPerWheel <<= 1;
}
return normalizedTicksPerWheel;
}
然后我們?cè)賮?lái)看看start方法，它是如何啟動(dòng)定時(shí)器的：
public void start() {
// 針對(duì)worker的狀態(tài)進(jìn)行switch
switch (WORKER_STATE_UPDATER.get(this)) {
// 如果是初始化
case WORKER_STATE_INIT:
// 如果能cas更新到開始狀態(tài)
if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
// 那就啟動(dòng)worker線程
workerThread.start();
}
break;
// 如果已經(jīng)處于啟動(dòng)监右，自然什么都不用做
case WORKER_STATE_STARTED:
break;
// 如果已經(jīng)shutdown边灭， 那也就進(jìn)入了非法狀態(tài)
case WORKER_STATE_SHUTDOWN:
throw new IllegalStateException("cannot be started once stopped");
default:
throw new Error("Invalid WorkerState");
}
// 這里需要同步等待worker線程啟動(dòng)并完成startTime初始化的工作
while (startTime == 0) {
try {
startTimeInitialized.await();
} catch (InterruptedException ignore) {
// Ignore - it will be ready very soon.
}
}
}
從這里我們看到，其實(shí)真正的邏輯全部到了worker里面健盒，這里只需要做一些基礎(chǔ)狀態(tài)判斷并等待work啟動(dòng)完畢即可绒瘦。
接下來(lái)就來(lái)到最后称簿，HashedWheelTimer實(shí)現(xiàn)的Timer接口的兩個(gè)函數(shù)了：
@Override
public Set<Timeout> stop() {
// 判斷當(dāng)前線程是否是worker線程，stop方法不能由TimerTask觸發(fā)惰帽，
// 否則后面的同步等待join操作就無(wú)法完成
if (Thread.currentThread() == workerThread) {
throw new IllegalStateException(
HashedWheelTimer.class.getSimpleName() +
".stop() cannot be called from " +
TimerTask.class.getSimpleName());
}
// 同樣的cas更新狀態(tài)
if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
// workerState can be 0 or 2 at this moment - let it always be 2.
if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
INSTANCE_COUNTER.decrementAndGet();
if (leak != null) {
boolean closed = leak.close(this);
assert closed;
}
}
return Collections.emptySet();
}
// 如果來(lái)到了這里憨降，說(shuō)明之前cas更新一切順利
try {
boolean interrupted = false;
// while循環(huán)持續(xù)中斷worker線程直到它醒悟它該結(jié)束了(有可能被一些耗時(shí)的操作耽誤了)
// 通過isAlive判斷worker線程是否已經(jīng)結(jié)束
while (workerThread.isAlive()) {
workerThread.interrupt();
try {
workerThread.join(100);
} catch (InterruptedException ignored) {
interrupted = true;
}
}
// 如果當(dāng)前線程被interrupt，就設(shè)置標(biāo)志位善茎，常規(guī)操作
if (interrupted) {
Thread.currentThread().interrupt();
}
} finally {
// 減少實(shí)例數(shù)
INSTANCE_COUNTER.decrementAndGet();
if (leak != null) {
boolean closed = leak.close(this);
assert closed;
}
}
// 返回還沒執(zhí)行的定時(shí)任務(wù)
return worker.unprocessedTimeouts();
}
@Override
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
// 基本的異常判斷
if (task == null) {
throw new NullPointerException("task");
}
if (unit == null) {
throw new NullPointerException("unit");
}
// 增加等待執(zhí)行的定時(shí)任務(wù)數(shù)
long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
// 如果超過最大就gg
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 + ")");
}
// 開啟定時(shí)器券册，還記得么频轿？
// 它的狀態(tài)判斷能夠很好的處理多次啟動(dòng)
// 并且還能幫我們做一下狀態(tài)判斷
start();
// 計(jì)算這個(gè)任務(wù)的deadline垂涯，startTime是worker線程開始的時(shí)間戳
long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
// 溢出保護(hù)
if (delay > 0 && deadline < 0) {
deadline = Long.MAX_VALUE;
}
// 直接創(chuàng)建一個(gè)HashedWheelTimeout句柄
HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
// 添加到等待執(zhí)行的任務(wù)隊(duì)列中（大家還記得上一節(jié)我們說(shuō)這是個(gè)MPSC隊(duì)列么）
timeouts.add(timeout);
// 返回這個(gè)句柄
return timeout;
}
這兩個(gè)方法很好的實(shí)現(xiàn)了本來(lái)應(yīng)該有的語(yǔ)義，這里注釋已經(jīng)把每個(gè)語(yǔ)句的含義解釋清楚了航邢，這里就不在多解釋耕赘。通過這里我們大概了解，其實(shí)大部分邏輯都在worker線程內(nèi)部膳殷，HashedWheelTimer只是做了一個(gè)比較好的開啟操骡、結(jié)束和插入任務(wù)的機(jī)制。接下來(lái)我們?cè)俚絯orker里面去看看赚窃。
worker
既然worker封裝的是工作線程的實(shí)現(xiàn)邏輯册招，那么我們肯定就的先來(lái)看看run方法了：
@Override
public void run() {
// 初始化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;
}
// 通知還等待在start方法上的線程，我worker初始化好了
startTimeInitialized.countDown();
do {
// 等待下一個(gè)tick
final long deadline = waitForNextTick();
if (deadline > 0) {
// 獲取下一個(gè)bucket的index勒极，即當(dāng)前tick mod bucket數(shù)量
int idx = (int) (tick & mask);
// 處理掉已取消的任務(wù)
processCancelledTasks();
// 獲取當(dāng)前要處理的bucket
HashedWheelBucket bucket =
wheel[idx];
// 將待處理的任務(wù)移動(dòng)到它該去的bucket去
transferTimeoutsToBuckets();
// 處理掉當(dāng)前bucket的所有到期定時(shí)任務(wù)
bucket.expireTimeouts(deadline);
// 遞增tick
tick++;
}
// 除非當(dāng)前的狀態(tài)不是started是掰，否則循環(huán)進(jìn)行下一個(gè)tick
} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
// 到了這里，說(shuō)明worker的狀態(tài)已經(jīng)不是started了辱匿，到了shutdown
// 那就得善后了键痛，遍歷所有的bucket，將還沒來(lái)得及處理的任務(wù)全部清理到unprocessedTimeouts中
for (HashedWheelBucket bucket: wheel) {
bucket.clearTimeouts(unprocessedTimeouts);
}
// 遍歷所有待處理并且還沒取消的任務(wù)匾七，添加到unprocessedTimeouts中
for (;;) {
HashedWheelTimeout timeout = timeouts.poll();
if (timeout == null) {
break;
}
if (!timeout.isCancelled()) {
unprocessedTimeouts.add(timeout);
}
}
// 處理所有的已取消的task絮短，防止內(nèi)存泄漏
// 其實(shí)我任務(wù)應(yīng)該在結(jié)束的時(shí)候先處理已經(jīng)取消的任務(wù)，這樣似乎好理解一些
// 不過我理解如果我這樣做有可能出現(xiàn)的問題是bucket里面會(huì)有任務(wù)殘留
// 特別是這個(gè)間隙其他線程還在不斷cancel任務(wù)昨忆，這里就不過多展開了
processCancelledTasks();
}
從run方法中我們可以看出丁频，這個(gè)方法定義了在每一個(gè)tick循環(huán)中需要執(zhí)行的任務(wù)以及他們的先后順序，還有在退出的時(shí)候worker線程需要做的善后工作邑贴。
那我們先來(lái)看waitForNextTick:
private long waitForNextTick() {
// 計(jì)算下一個(gè)tick的deadline
long deadline = tickDuration * (tick + 1);
// 循環(huán)直到當(dāng)前時(shí)間來(lái)到了下一個(gè)tick
for (;;) {
// 計(jì)算當(dāng)前時(shí)間
final long currentTime = System.nanoTime() - startTime;
// 計(jì)算需要sleep的毫秒數(shù)
long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
// 如果sleep的毫秒數(shù)小于等于0
if (sleepTimeMs <= 0) {
// 特殊判斷 這里說(shuō)實(shí)話我沒咋看懂
if (currentTime == Long.MIN_VALUE) {
return -Long.MAX_VALUE;
} else {
// 無(wú)需等待席里，直接返回
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
// 處理一些windows才有的jvm bug
if (PlatformDependent.isWindows()) {
sleepTimeMs = sleepTimeMs / 10 * 10;
}
// 嘗試sleep到下個(gè)tick的deadline
try {
Thread.sleep(sleepTimeMs);
} catch (InterruptedException ignored) {
// 如果發(fā)現(xiàn)已經(jīng)被shutdown了，則返回MIN_VALUE痢缎，可以讓run快速處理
if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
return Long.MIN_VALUE;
}
}
}
}
這個(gè)方法的核心目標(biāo)其實(shí)就是讓worker線程能夠等夠足夠多的時(shí)間到下一個(gè)tick胁勺，并做了一些特殊處理。接下來(lái)我們看看worker線程如何分發(fā)任務(wù)到bucket的：
private void transferTimeoutsToBuckets() {
// 最多一次轉(zhuǎn)移100000個(gè)待分發(fā)定時(shí)任務(wù)到它們對(duì)應(yīng)的bucket內(nèi)
// 不然如果有一個(gè)線程一直添加定時(shí)任務(wù)就能讓工作線程活生生餓死
for (int i = 0; i < 100000; i++) {
// 獲取一個(gè)定時(shí)任務(wù)
HashedWheelTimeout timeout = timeouts.poll();
if (timeout == null) {
// all processed
break;
}
// 如果已經(jīng)取消就不管了
if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
// Was cancelled in the meantime.
continue;
}
// 計(jì)算從worker線程開始運(yùn)行算起要經(jīng)過多少個(gè)tick才能到這個(gè)任務(wù)
long calculated = timeout.deadline / tickDuration;
// 計(jì)算這個(gè)任務(wù)要經(jīng)過多少圈
timeout.remainingRounds = (calculated - tick) / wheel.length;
// 如果這個(gè)任務(wù)我們?nèi)⊥砹硕揽酰蔷妥屗谶@個(gè)tick執(zhí)行
final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
// 計(jì)算他應(yīng)該到的bucket的index
int stopIndex = (int) (ticks & mask);
// 指派過去即可
HashedWheelBucket bucket = wheel[stopIndex];
bucket.addTimeout(timeout);
}
}
這里代碼也比較簡(jiǎn)單署穗，我就不過多解釋了寥裂，需要注意的就是一次最多只會(huì)分發(fā)100000個(gè)任務(wù)，防止worker沒法做其他的事情案疲。接著我們?cè)賮?lái)看看worker如何處理已取消的任務(wù)的：
private void processCancelledTasks() {
// 畢竟取消的任務(wù)占極少數(shù)封恰，所以這里就沒有個(gè)數(shù)限制了
for (;;) {
// 取出一個(gè)取消的任務(wù)
HashedWheelTimeout timeout = cancelledTimeouts.poll();
if (timeout == null) {
// all processed
break;
}
try {
// 這里實(shí)際上是將它從它所屬的bucket的雙向鏈表中刪除，這里后面會(huì)看到
timeout.remove();
} catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("An exception was thrown while process a cancellation task", t);
}
}
}
}
同樣也比較簡(jiǎn)單褐啡，worker就分析完了诺舔。接著我們?cè)偃ashedWheelBucket，看看時(shí)間輪上每一個(gè)單元做了什么备畦，因?yàn)閣orker線程run方法中最終處理定時(shí)任務(wù)實(shí)際上是調(diào)用了bucket.expireTimeouts(deadline);
HashedWheelBucket
public void expireTimeouts(long deadline) {
// 獲取雙向鏈表的頭
HashedWheelTimeout timeout = head;
// 從頭到尾處理所有的任務(wù)
while (timeout != null) {
HashedWheelTimeout next = timeout.next;
// 如果剩余輪數(shù)小于0 說(shuō)明需要馬上執(zhí)行
if (timeout.remainingRounds <= 0) {
// 將它從當(dāng)前鏈表中移除
next = remove(timeout);
if (timeout.deadline <= deadline) {
// 執(zhí)行timeout的邏輯
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()) {
// 如果已經(jīng)被取消低飒，同樣remove掉
next = remove(timeout);
} else {
// 否則將他們的輪數(shù)減一
timeout.remainingRounds --;
}
// 繼續(xù)鏈表的遍歷
timeout = next;
}
}
這里實(shí)際上就是遍歷整個(gè)鏈表，將應(yīng)該執(zhí)行的執(zhí)行掉懂盐，然后其余的維護(hù)他們的狀態(tài)褥赊，保證后續(xù)的執(zhí)行。
HashedWheelBucket中其他的方法基本上都是關(guān)于雙向鏈表的操作莉恼，這里就不在贅述了拌喉，需要注意的是，由于所有的工作都在worker線程中進(jìn)行俐银，因此基本不需要任何線程安全機(jī)制尿背，保證了高性能。
最后捶惜，我們還是來(lái)到了HashedWheelTimeout田藐，看看一個(gè)定時(shí)任務(wù)內(nèi)部需要有哪些處理。
HashedWheelTimeout
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);
}
}
}
大家應(yīng)該還記得之前bucket.expireTimeouts(deadline);里面就是調(diào)用了timeout.expire(); 來(lái)執(zhí)行具體的業(yè)務(wù)邏輯售躁，而這里我們也看到坞淮，除了一些基本的狀態(tài)判斷，就是直接執(zhí)行了task中的run方法陪捷。所以這里需要大家注意的就是task的run方法一定慎重放置任何有耗時(shí)可能的操作回窘，不然就會(huì)導(dǎo)致HashedWheelTimer中worker線程被長(zhǎng)時(shí)間占用，其他任務(wù)得不到執(zhí)行或者無(wú)法準(zhǔn)時(shí)執(zhí)行市袖，最終導(dǎo)致性能和正確性下降啡直。
其他的方法也太過簡(jiǎn)單，這里就不提及苍碟，有興趣的同學(xué)可以直接查看源碼酒觅。
結(jié)語(yǔ)
這篇文章針對(duì)netty中的定時(shí)任務(wù)處理機(jī)制HashedWheelTimer的實(shí)現(xiàn)以及背后原理進(jìn)行了一定程度的闡述和說(shuō)明，希望能對(duì)大家理解有所幫助微峰。
