Netty时间轮_java netty

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
</dependency>

 编写测试:

CountDownLatch countDownLatch = new CountDownLatch(1);
HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 16);
log.info("current timestamp={}",System.currentTimeMillis());
timer.newTimeout((timeout) -> { 

log.info("task execute,current timestamp={}",System.currentTimeMillis());
countDownLatch.countDown();
}, 2000, TimeUnit.MILLISECONDS);
countDownLatch.await();
timer.stop();

public interface Timer { 

Timeout newTimeout(TimerTask task, long delay, TimeUnit unit);
Set<Timeout> stop();
}

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

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

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

HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 16);

public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) { 

this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
}
public HashedWheelTimer(
ThreadFactory threadFactory,
long tickDuration, TimeUnit unit, int ticksPerWheel) { 

this(threadFactory, tickDuration, unit, ticksPerWheel, true);
}

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

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

private int normalizeTicksPerWheel(int ticksPerWheel) { 

int n = ticksPerWheel - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
// 这里1073741824 = 2^30,防止溢出
return (n < 0) ? 1 : (n >= 1073741824) ? 1073741824 : n + 1;
}

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

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

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

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

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

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