Afred's Blog / netty

public HashedWheelTimer(
            ThreadFactory threadFactory,
            long tickDuration, TimeUnit unit, int ticksPerWheel)

// constructor start 
  // Normalize ticksPerWheel to power of two and initialize the wheel.
  // 对齐，生成的格子数量为2的n次方
  wheel = createWheel(ticksPerWheel);
  // 由于wheel.length等于2的n次方，mask的二进制低位全是1，常规操作
  mask = wheel.length - 1;

  // Convert tickDuration to nanos.
  this.tickDuration = unit.toNanos(tickDuration);

  // Prevent overflow.
  if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
      throw new IllegalArgumentException(String.format(
              "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
              tickDuration, Long.MAX_VALUE / wheel.length));
  }

  // 创建Worker线程，此时并没有启动，如果此时没有任务，线程空转会有一定的资源浪费
  workerThread = threadFactory.newThread(worker);
// constructor end


private static Set<HashedWheelTimeout>[] 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);
    Set<HashedWheelTimeout>[] wheel = new Set[ticksPerWheel];
    for (int i = 0; i < wheel.length; i ++) {
        wheel[i] = Collections.newSetFromMap(
                PlatformDependent.<HashedWheelTimeout, Boolean>newConcurrentHashMap());
    }
    return wheel;
}

// 对齐格子数量，稍大于ticksPerWheel且是2的n次方
private static int normalizeTicksPerWheel(int ticksPerWheel) {
    int normalizedTicksPerWheel = 1;
    while (normalizedTicksPerWheel < ticksPerWheel) {
        normalizedTicksPerWheel <<= 1;
    }
    return normalizedTicksPerWheel;
}

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

task : 任务的实现
delay : 指定延迟时间
unit : 延迟时间的时间单位

// newTimeout 函数
// 理论上的任务执行时间和时间轮启动时间之间的时间差
long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
  // Add the timeout to the wheel.
  HashedWheelTimeout timeout;
  lock.readLock().lock();
  try {
  // 在读锁之后调用构造函数
      timeout = new HashedWheelTimeout(task, deadline);
      if (workerState.get() == WORKER_STATE_SHUTDOWN) {
          throw new IllegalStateException("Cannot enqueue after shutdown");
      }
      wheel[timeout.stopIndex].add(timeout);
  } finally {
      lock.readLock().unlock();
  }
  
 // HashedWheelTimeout 类
 // 任务状态标志
private static final int ST_INIT = 0;
private static final int ST_CANCELLED = 1;
private static final int ST_EXPIRED = 2;


private final TimerTask task;	// 具体任务
final long deadline;	// 同上面的deadline
final int stopIndex;	// 槽位
volatile long remainingRounds;	// 剩余轮次，有可能当前任务需要时间轮回多次，如果大于0，则本次不执行，需要使用volatile保证可见性
private final AtomicInteger state = new AtomicInteger(ST_INIT);

HashedWheelTimeout(TimerTask task, long deadline) {
    this.task = task;
    this.deadline = deadline;
	// 计算槽位
   long calculated = deadline / tickDuration;
   // 已经过期的任务，存入当前槽位直接，方便worker在方法调用完后执行（添加和执行分别使用读写锁，可能当前线程在读写锁竞争失败，槽位calculated的任务已经执行完成，此时将任务添加到当前槽位tick中）
    final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
    stopIndex = (int) (ticks & mask);
    // 可能小于等于0，如果是小于等于0，Worker会在当前tick中执行
    remainingRounds = (calculated - tick) / wheel.length;
}

// run 方法
do {
    final long deadline = waitForNextTick();
    if (deadline > 0) {
        fetchExpiredTimeouts(expiredTimeouts, deadline);
        notifyExpiredTimeouts(expiredTimeouts);
    }
} while (workerState.get() == WORKER_STATE_STARTED);

// waitForNextTick 方法
/**
    * calculate goal nanoTime from startTime and current tick number,
    * then wait until that goal has been reached.
    * @return Long.MIN_VALUE if received a shutdown request,
 * current time otherwise (with Long.MIN_VALUE changed by +1)
 */
 // sleep，直到下一个tick到来
private long waitForNextTick() {
    long deadline = tickDuration * (tick + 1);

    for (;;) {
        final long currentTime = System.nanoTime() - startTime;
        //计算需要sleep的时间, 之所以加9999999后再除10000000, 是因为保证为10毫秒的倍数.
        long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;

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

        try {
            Thread.sleep(sleepTimeMs);
                } catch (InterruptedException e) {
            //当调用Timer.stop时, 退出
            if (workerState.get() == WORKER_STATE_SHUTDOWN) {
                return Long.MIN_VALUE;
            }
        }
    }

private void fetchExpiredTimeouts(
            List<HashedWheelTimeout> expiredTimeouts, long deadline) {

    // Find the expired timeouts and decrease the round counter
        // if necessary.  Note that we don't send the notification
        // immediately to make sure the listeners are called without
        // an exclusive lock.
        lock.writeLock().lock();
        try {
            fetchExpiredTimeouts(expiredTimeouts, wheel[(int) (tick & mask)].iterator(), deadline);
        } finally {
            // Note that the tick is updated only while the writer lock is held,
            // so that newTimeout() and consequently new HashedWheelTimeout() never see an old value
            // while the reader lock is held.
            tick ++;
            lock.writeLock().unlock();
        }
    }

private void fetchExpiredTimeouts(
            List<HashedWheelTimeout> expiredTimeouts,
            Iterator<HashedWheelTimeout> i, long deadline) {

    while (i.hasNext()) {
            HashedWheelTimeout timeout = i.next();
            if (timeout.remainingRounds <= 0) {
                i.remove();
                if (timeout.deadline <= deadline) {
                    expiredTimeouts.add(timeout);
                } else {
                    // The timeout was placed into a wrong slot. This should never happen.
                    throw new Error(String.format(
                            "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                }
            } else {
                timeout.remainingRounds --;
            }
        }
    }

private void notifyExpiredTimeouts(
            List<HashedWheelTimeout> expiredTimeouts) {
        // Notify the expired timeouts.
        for (int i = expiredTimeouts.size() - 1; i >= 0; i --) {
        // 同步轮训调用，根据下文贴出的HashedWheelTimeout#expire方法，如果TimerTask的run方法是耗时操作，则会影响Worker线程的执行，所以应该在TimerTask中封装一层，将业务逻辑扔到业务线程池中执行
            expiredTimeouts.get(i).expire();
        }

    // Clean up the temporary list.
        expiredTimeouts.clear();
    }
    
 // HashedWheelTimeout 类
 public void expire() {
    if (!state.compareAndSet(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);
        }
    }
}

@Override
public boolean cancel() {
    if (!state.compareAndSet(ST_INIT, ST_CANCELLED)) {
        return false;
    }

    wheel[stopIndex].remove(this);
    return true;
}