关于java:追踪解析-Netty-的-HashedWheelTimer-源码

41次阅读

共计 6125 个字符，预计需要花费 16 分钟才能阅读完成。

零 后期筹备

0 FBI WARNING

文章异样啰嗦且绕弯。

1 版本

JDK 版本 : Adoptopenjdk 14.0.1

IDE : idea 2020.1

Netty 版本 : netty-all 4.1.46.Final

2 HashedWheelTimer 简介

HashedWheelTimer 是 Netty 中实现提早工作的工具类。

3 Demo


import io.netty.util.HashedWheelTimer;

import io.netty.util.Timeout;

import io.netty.util.TimerTask;

import java.util.concurrent.TimeUnit;

/**

* netty 工夫轮 demo

*/

public class TimeWheel {public static void main(String[] args) {

// 创立工夫轮对象

HashedWheelTimer wheel = new HashedWheelTimer();

// 打印以后工夫

System.out.println(System.currentTimeMillis());

// 创立一个提早工作，用于打印

wheel.newTimeout(new TimerTask() {

@Override

public void run(Timeout timeout) throws Exception {System.out.println("test print" + System.currentTimeMillis());

}

},1000, TimeUnit.MILLISECONDS);

// 线程休眠，避免线程退出

try {Thread.sleep(1000000000L);

} catch (InterruptedException e) {e.printStackTrace();

}

}

}

4 创立 HashedTWheelTimer

4.1 结构器

上述 demo 中创立 HashedTWheelTimer：


HashedWheelTimer wheel = new HashedWheelTimer();

追踪它的实现：


/**

* 其它所有结构器都最终调用到这个结构器

* threadFactory - 线程工厂，默认是 Executors 中的 defaultThreadFactory

* tickDuration - 工夫轮两次 tick 之间的间隔时间，默认值 100

* unit - tick 的工夫单位，默认为 毫秒

* ticksPerWheel - 一轮 tick 的数量

* leakDetection - 用于优雅敞开的援用对象，默认 true

* maxPendingTimeouts - 最大工作数，默认 -1，即为不限度

**/

public HashedWheelTimer(

ThreadFactory threadFactory,

long tickDuration,

TimeUnit unit,

int ticksPerWheel,

boolean leakDetection,

long maxPendingTimeouts) {

// 做局部参数的非空和合规验证

ObjectUtil.checkNotNull(threadFactory, "threadFactory");

ObjectUtil.checkNotNull(unit, "unit");

ObjectUtil.checkPositive(tickDuration, "tickDuration");

ObjectUtil.checkPositive(ticksPerWheel, "ticksPerWheel");

// wheel 是一个 HashedWheelBucket 数组，是工作存储器，具体见 4.2

wheel = createWheel(ticksPerWheel);

mask = wheel.length - 1;

// 获取实在的 tick 工夫，并测验合法性

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

}

// MILLISECOND_NANOS = 1000000

// duration 不得小于 1000000

if (duration < MILLISECOND_NANOS) {logger.warn("Configured tickDuration {} smaller then {}, using 1ms.",tickDuration, MILLISECOND_NANOS);

this.tickDuration = MILLISECOND_NANOS;

} else {this.tickDuration = duration;}

// 工作线程，具体见 4.3

workerThread = threadFactory.newThread(worker);

// 判断是否须要创立一个用于优雅敞开的弱援用对象

leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;

// 最大工作数

this.maxPendingTimeouts = maxPendingTimeouts;

// 工夫轮在一个 jvm 过程中的最大初始化数量不得大于 64 个，不然的话会在此处报错

if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&

WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {reportTooManyInstances();

}

}

4.2 HashedWheelBucket

wheel 是一个 HashedWheelBucket 数组对象：


private final HashedWheelBucket[] wheel;

wheel 的初始化在 createWheel(…) 办法中进行：


// HashedWheelBucket.class

private static HashedWheelBucket[] createWheel(int ticksPerWheel) {

// ticks 是在创立 HashedWheelBucket 的时候输出的分片数

// 分片数不可小于 0，或者大于 1073741824

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

}

// ticks 必须是 2 的倍数，所以此处会对使用者传入的 ticksPerWheel 进行修改

ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);

// 创立数组并初始化

HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];

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

// 返回

return wheel;

}

HashedWheelBucket 是 HashedWheelTimer 的外部类：


private static final class HashedWheelBucket {

// HashedWheelBucket 实质上是 HashedWheelTimeout 的链表，而 HashedWheelTimeout 实质上是要执行的工作

private HashedWheelTimeout head;

private HashedWheelTimeout tail;

// ...

}

当使用者存入一个工作的时候，会调用到 HashedWheelBucket 的 addTimeout(…) 办法：


// 实质上是将工作放入到链表中

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;

}

}

而取出工作要生产的时候，调用 pollTimeout() 办法：


// 实质上是将工作从链表中拿出

private HashedWheelTimeout pollTimeout() {

HashedWheelTimeout head = this.head;

if (head == null) {return null;}

HashedWheelTimeout next = head.next;

if (next == null) {tail = this.head = null;} else {

this.head = next;

next.prev = null;

}

head.next = null;

head.prev = null;

head.bucket = null;

return head;

}

4.3 workerThread

workerThread 是一个线程对象：


private final Thread workerThread;

调用线程工厂的 newThread 办法实现初始化：


workerThread = threadFactory.newThread(worker);

worker 对象是一个 Worker 对象。Worker 是 HashedWheelTimer 的外部类，是 Runnable 的实现类：


private final class Worker implements Runnable {private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();

private long tick;

@Override

public void run() {

// 记录开始工夫

startTime = System.nanoTime();

if (startTime == 0) {startTime = 1;}

// CountDownLatch，确保先实现初始化之后才会塞入工作

startTimeInitialized.countDown();

do {

// 在这个办法中线程会休眠，直到下一个 tick 的工夫

final long deadline = waitForNextTick();

if (deadline > 0) {int idx = (int) (tick & mask);

// 清理曾经被勾销的 task

processCancelledTasks();

// 选定以后要解决的 bucket

HashedWheelBucket bucket = wheel[idx];

// 将贮存在队列里的工作搁置到对应的 bucket 里

transferTimeoutsToBuckets();

// 执行以后 bucket 里所有的工作

bucket.expireTimeouts(deadline);

// 切换到下一个 tick

tick++;

}

} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);

// 后续代码用以优雅敞开

for (HashedWheelBucket bucket: wheel) {bucket.clearTimeouts(unprocessedTimeouts);

}

for (;;) {HashedWheelTimeout timeout = timeouts.poll();

if (timeout == null) {break;}

if (!timeout.isCancelled()) {unprocessedTimeouts.add(timeout);

}

}

processCancelledTasks();}

// 略 ...

}

5 增加工作

回到 demo 的增加工作的代码：


wheel.newTimeout(new TimerTask() {

@Override

public void run(Timeout timeout) throws Exception {System.out.println("test print" + System.currentTimeMillis());

}

},1000, TimeUnit.MILLISECONDS);

追踪 newTimeout(…) 办法：


// HashedWheelTimer.java

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

// 非空验证

ObjectUtil.checkNotNull(task, "task");

ObjectUtil.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();

// 该当被执行的延迟时间

long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;

// 时钟回拨的状况

if (delay > 0 && deadline < 0) {deadline = Long.MAX_VALUE;}

// 将 task 包装成一个 timeout 对象

HashedWheelTimeout timeout = new HashedWheelTimeout(this,task,deadline);

// 放到队列里

timeouts.add(timeout);

return timeout;

}

本文仅为集体的学习笔记，可能存在谬误或者表述不清的中央，有缘补充

正文完
java
发表至： java
2020-09-08
 0
关于java:Ztree实现增删改查
关于java:后端面经JavaHashMap详解
关于java:2020年以后IT行业还能否继续那么吃香
关于酷工作:酷工作丨涂鸦智能招研发工程师架构师技术专家华沁智能交通多岗位招聘氢舟科技招前端工程师