任务调度：时间轮算法经典案例解析及应用实现

private final TaskQueue queue = new TaskQueue();//基于数组实现的优先队列，用作存放任务
private final TimerThread thread = new TimerThread(queue);//执行任务线程
public Timer(String name) {        thread.setName(name);        thread.start(); //构造时默认启动线程    }

class TaskQueue {    private TimerTask[] queue = new TimerTask[128];    void add(TimerTask task) {        // Grow backing store if necessary        if (size + 1 == queue.length)            queue = Arrays.copyOf(queue, 2*queue.length); //扩容
        queue[++size] = task; //先将任务添加到数组最后面        fixUp(size); //调整堆    }    private void fixUp(int k) { //时间复杂度为O(logn)        while (k > 1) {            int j = k >> 1;            if (queue[j].nextExecutionTime <= queue[k].nextExecutionTime)//通过任务执行时间对比，调整顺序                break;            TimerTask tmp = queue[j];  queue[j] = queue[k]; queue[k] = tmp;            k = j;        }    }  /**     * Return the "head task" of the priority queue.  (The head task is an     * task with the lowest nextExecutionTime.)     */    TimerTask getMin() {        return queue[1]; //返回最接近执行时间的任务    }     //.......}

public void run() {        try {            mainLoop();//无异常捕获        } finally {            // Someone killed this Thread, behave as if Timer cancelled            synchronized(queue) {                newTasksMayBeScheduled = false;                queue.clear();  // Eliminate obsolete references            }        }    }
    /**     * The main timer loop.  (See class comment.)     */    private void mainLoop() {        while (true) {            try {                TimerTask task;                boolean taskFired;                synchronized(queue) {                    // Wait for queue to become non-empty                    while (queue.isEmpty() && newTasksMayBeScheduled)                        queue.wait();                    if (queue.isEmpty())                        break; // Queue is empty and will forever remain; die
                    // Queue nonempty; look at first evt and do the right thing                    long currentTime, executionTime;                    task = queue.getMin(); //获取任务                    synchronized(task.lock) {                        if (task.state == TimerTask.CANCELLED) { //取消泽移除并继续循环                            queue.removeMin();                            continue;  // No action required, poll queue again                        }                        currentTime = System.currentTimeMillis();                        executionTime = task.nextExecutionTime;                        if (taskFired = (executionTime<=currentTime)) { //执行时间到了                            if (task.period == 0) { // 不是周期任务                                queue.removeMin(); //移除任务                                task.state = TimerTask.EXECUTED;//变更任务状态为已执行                            } else { // 周期任务，更新时间为下次执行时间                                queue.rescheduleMin(                                  task.period<0 ? currentTime   - task.period                                                : executionTime + task.period);                            }                        }                    }                    if (!taskFired) // 还未到达执行时间等待                        queue.wait(executionTime - currentTime);                }                if (taskFired)  // 执行任务，无异常捕获                    task.run();            } catch(InterruptedException e) {            }        }    }

/** * ....... * <p>Java 5.0 introduced the {@code java.util.concurrent} package and * one of the concurrency utilities therein is the {@link * java.util.concurrent.ScheduledThreadPoolExecutor * ScheduledThreadPoolExecutor} which is a thread pool for repeatedly * executing tasks at a given rate or delay.  It is effectively a more * versatile replacement for the {@code Timer}/{@code TimerTask} * combination, as it allows multiple service threads, accepts various * time units, and doesn't require subclassing {@code TimerTask} (just * implement {@code Runnable}).  Configuring {@code * ScheduledThreadPoolExecutor} with one thread makes it equivalent to * {@code Timer}. * ...... */public class Timer {  //。。。。}

public class ScheduledThreadPoolExecutor        extends ThreadPoolExecutor        implements ScheduledExecutorService {  // ...  public ScheduledThreadPoolExecutor(int corePoolSize) {        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,              new DelayedWorkQueue());    }private class ScheduledFutureTask<V>            extends FutureTask<V> implements RunnableScheduledFuture<V> {//...}}

private class ScheduledFutureTask<V>            extends FutureTask<V> implements RunnableScheduledFuture<V> {//...   public void run() {            boolean periodic = isPeriodic();            if (!canRunInCurrentRunState(periodic))                cancel(false);            else if (!periodic)                ScheduledFutureTask.super.run(); //如果不是周期任务，直接run            else if (ScheduledFutureTask.super.runAndReset()) {//周期性任务,执行并重置任务状态为NEW                setNextRunTime();//先设置下次任务执行时间                reExecutePeriodic(outerTask); //任务重新入队列            }        }//...}

static class DelayedWorkQueue extends AbstractQueue<Runnable>        implements BlockingQueue<Runnable> {
 public RunnableScheduledFuture<?> take() throws InterruptedException {            final ReentrantLock lock = this.lock;            lock.lockInterruptibly();            try {                for (;;) {                    RunnableScheduledFuture<?> first = queue[0];                    if (first == null)                        available.await(); //如果队列中没有任务，等待                    else {                        long delay = first.getDelay(NANOSECONDS);                        if (delay <= 0)//有任务并且到执行时间了                            return finishPoll(first);                        first = null; // don't retain ref while waiting                        if (leader != null)                            available.await();                        else {                            Thread thisThread = Thread.currentThread();                            leader = thisThread;                            try {                                available.awaitNanos(delay);//有任务，没有到执行时间                            } finally {                                if (leader == thisThread)                                    leader = null;                            }                        }                    }                }            } finally {                if (leader == null && queue[0] != null)                    available.signal();                lock.unlock();            }        }
}

//元素必须实现Delayed接口，也实现了阻塞队列public class DelayQueue<E extends Delayed> extends AbstractQueue<E>    implements BlockingQueue<E> {
    private final transient ReentrantLock lock = new ReentrantLock();    private final PriorityQueue<E> q = new PriorityQueue<E>();//优先队列，       public E take() throws InterruptedException {        final ReentrantLock lock = this.lock;        lock.lockInterruptibly();        try {            for (;;) {                E first = q.peek();                if (first == null)                    available.await();                else {                    long delay = first.getDelay(NANOSECONDS);                    if (delay <= 0) //小于等于0，时间到了                        return q.poll();                    first = null; // don't retain ref while waiting                    if (leader != null)                        available.await();//没有抢到leader的线程进入等待，避免大量唤醒操作                    else {                        Thread thisThread = Thread.currentThread();                        leader = thisThread;                        try {                            available.awaitNanos(delay);//leader线程，在等待一定时间后再次尝试获取                        } finally {                            if (leader == thisThread)//重置leader                                leader = null;                        }                    }                }            }        } finally {            if (leader == null && q.peek() != null)                available.signal();            lock.unlock();        }    }    //...}//继承了Comparablepublic interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);}

public class HashedWheelTimer implements Timer {//... public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {        this(threadFactory, tickDuration, unit, 512);    } public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts) {        this.worker = new HashedWheelTimer.Worker();        this.workerState = 0;        this.startTimeInitialized = new CountDownLatch(1);        this.timeouts = PlatformDependent.newMpscQueue();        this.cancelledTimeouts = PlatformDependent.newMpscQueue();        this.pendingTimeouts = new AtomicLong(0L);        if (threadFactory == null) {            throw new NullPointerException("threadFactory");        } else if (unit == null) {            throw new NullPointerException("unit");        } else if (tickDuration <= 0L) {            throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);        } else if (ticksPerWheel <= 0) {            throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);        } else {            this.wheel = createWheel(ticksPerWheel);//默认512的槽数量            this.mask = this.wheel.length - 1;//mask设计提高了“确定槽位下标”的性能，通过限制wheel.length为2的N次方，            this.tickDuration = unit.toNanos(tickDuration);//时间精度，即每个槽位的时间跨度            if (this.tickDuration >= 9223372036854775807L / (long)this.wheel.length) {                throw new IllegalArgumentException(String.format("tickDuration: %d (expected: 0 < tickDuration in nanos < %d", tickDuration, 9223372036854775807L / (long)this.wheel.length));            } else {                this.workerThread = threadFactory.newThread(this.worker);//工作线程                this.leak = !leakDetection && this.workerThread.isDaemon() ? null : leakDetector.track(this);                this.maxPendingTimeouts = maxPendingTimeouts;                if (INSTANCE_COUNTER.incrementAndGet() > 64 && WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {                    reportTooManyInstances();                }            }        }    }//...
//双向链表的bucketprivate static final class HashedWheelBucket {        private HashedWheelTimer.HashedWheelTimeout head;        private HashedWheelTimer.HashedWheelTimeout tail;//...}//链表元素，任务private static final class HashedWheelTimeout implements Timeout {  HashedWheelTimer.HashedWheelTimeout next;        HashedWheelTimer.HashedWheelTimeout prev;        HashedWheelTimer.HashedWheelBucket bucket;//...}}

  public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {        if (task == null) {            throw new NullPointerException("task");        } else if (unit == null) {            throw new NullPointerException("unit");        } else {            long pendingTimeoutsCount = this.pendingTimeouts.incrementAndGet();            if (this.maxPendingTimeouts > 0L && pendingTimeoutsCount > this.maxPendingTimeouts) {                this.pendingTimeouts.decrementAndGet();                throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending timeouts (" + this.maxPendingTimeouts + ")");            } else {                this.start();//启动工作线程，内部有判断状态                long deadline = System.nanoTime() + unit.toNanos(delay) - this.startTime;//计算延迟时间                HashedWheelTimer.HashedWheelTimeout timeout = new HashedWheelTimer.HashedWheelTimeout(this, task, deadline);//创建任务对象                this.timeouts.add(timeout);//添加到队列中，这边标记问题，没看到添加到时间轮中？                return timeout;            }        }    }

private final class Worker implements Runnable {//工作线程//...        public void run() {            HashedWheelTimer.this.startTime = System.nanoTime();//启动时间            if (HashedWheelTimer.this.startTime == 0L) {                HashedWheelTimer.this.startTime = 1L;            }            HashedWheelTimer.this.startTimeInitialized.countDown();
            int idx;            HashedWheelTimer.HashedWheelBucket bucket;            do {                long deadline = this.waitForNextTick();//等待执行任务时间到来                if (deadline > 0L) {                     idx = (int)(this.tick & (long)HashedWheelTimer.this.mask);//或者槽位下标                    this.processCancelledTasks();//先处理取消了的任务                    bucket = HashedWheelTimer.this.wheel[idx];//获取对应的槽                    this.transferTimeoutsToBuckets();//将添加到任务队列中的任务添加到时间轮中的槽中                    bucket.expireTimeouts(deadline);//处理时间到了的任务                    ++this.tick;//移动槽位                }            } while(HashedWheelTimer.WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == 1);        //...        }

private long waitForNextTick() {            long deadline = HashedWheelTimer.this.tickDuration * (this.tick + 1L);//根据时间精度，算出需要下一次的检查时间            while(true) {                long currentTime = System.nanoTime() - HashedWheelTimer.this.startTime;                long sleepTimeMs = (deadline - currentTime + 999999L) / 1000000L;                if (sleepTimeMs <= 0L) {//不用睡了，时间已经到了，直接执行                    if (currentTime == -9223372036854775808L) {//溢出了兜底？                        return -9223372036854775807L;                    }
                    return currentTime;                }                if (PlatformDependent.isWindows()) {//windows下的bug，sleep的时间需是10的整数倍                    sleepTimeMs = sleepTimeMs / 10L * 10L;                }                try {                    Thread.sleep(sleepTimeMs);//等待时间的到来                } catch (InterruptedException var8) {                    if (HashedWheelTimer.WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == 2) {                        return -9223372036854775808L;                    }                }            }        }

 private void transferTimeoutsToBuckets() {            for(int i = 0; i < 100000; ++i) {//设置了一次性处理10w个任务入槽，可能考虑这块处理太多会影响后续任务的处理？                HashedWheelTimer.HashedWheelTimeout timeout = (HashedWheelTimer.HashedWheelTimeout)HashedWheelTimer.this.timeouts.poll();//从队列中拿出任务                if (timeout == null) {                    break;                }                if (timeout.state() != 1) {                    long calculated = timeout.deadline / HashedWheelTimer.this.tickDuration;                    timeout.remainingRounds = (calculated - this.tick) / (long)HashedWheelTimer.this.wheel.length;//计算轮数                    long ticks = Math.max(calculated, this.tick);//如果时间已经过了，放到当前即将执行的槽位中                    int stopIndex = (int)(ticks & (long)HashedWheelTimer.this.mask);//计算槽位下标                    HashedWheelTimer.HashedWheelBucket bucket = HashedWheelTimer.this.wheel[stopIndex];                    bucket.addTimeout(timeout);//入槽                }            }        }

  public void expireTimeouts(long deadline) {            HashedWheelTimer.HashedWheelTimeout next;            for(HashedWheelTimer.HashedWheelTimeout timeout = this.head; timeout != null; timeout = next) {//循环链表                next = timeout.next;                if (timeout.remainingRounds <= 0L) {//处理轮数为0的任务                    next = this.remove(timeout);                    if (timeout.deadline > deadline) {                        throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));                    }                    timeout.expire();//过期任务，就是执行任务的run防范                } else if (timeout.isCancelled()) {                    next = this.remove(timeout);//任务取消了                } else {                    --timeout.remainingRounds;//轮数减一                }            }        }

@nonthreadsafeprivate[timer] class TimingWheel(tickMs: Long, wheelSize: Int, startMs: Long, taskCounter: AtomicInteger, queue: DelayQueue[TimerTaskList]) {//queue是一个TimerTaskList的延迟队列,每个槽的TimerTaskList都被加到这个延迟队列中，expireTime最小的槽会排在队列的最前面,此处要注意，这是kafka实现的一个特殊地方  private[this] val interval = tickMs * wheelSize //该层级时间轮的时间跨度  private[this] val buckets = Array.tabulate[TimerTaskList](wheelSize) { _ => new TimerTaskList(taskCounter) }
  private[this] var currentTime = startMs - (startMs % tickMs) //起始时间(startMs)都设置为创建此层时间轮时前面一轮的currentTime，每一层的currentTime都必须是tickMs的整数倍  @volatile private[this] var overflowWheel: TimingWheel = null

  //当时间溢出时，需要新增上一级时间轮，通过overflowWheel引用上一级侧时间轮  private[this] def addOverflowWheel(): Unit = {    synchronized {      if (overflowWheel == null) {        overflowWheel = new TimingWheel(          tickMs = interval,          wheelSize = wheelSize,          startMs = currentTime,          taskCounter = taskCounter,          queue        )      }    }  }
//添加任务  def add(timerTaskEntry: TimerTaskEntry): Boolean = {    val expiration = timerTaskEntry.expirationMs//获取任务延迟时间    if (timerTaskEntry.cancelled) {      false//取消    } else if (expiration < currentTime + tickMs) {      false      // 过期    } else if (expiration < currentTime + interval) {      //判断当前时间轮所能表示的时间范围是否可以容纳该任务      //计算槽位      val virtualId = expiration / tickMs      val bucket = buckets((virtualId % wheelSize.toLong).toInt)      bucket.add(timerTaskEntry)//将任务添加到槽位链表中      if (bucket.setExpiration(virtualId * tickMs)) {//设置expiretime        //将槽位的TimerTaskList添加到延迟队列中        queue.offer(bucket)      }      true    } else {      //到此处，说明时间溢出了，应该讲任务丢给父时间轮处理      if (overflowWheel == null) addOverflowWheel()//判断上一级时间轮是否存在，不存在创建      overflowWheel.add(timerTaskEntry)//调用上一级时间轮的添加任务方法，递归处理，直至添加成功    }  }
  //时间轮的推进  def advanceClock(timeMs: Long): Unit = {    if (timeMs >= currentTime + tickMs) {      currentTime = timeMs - (timeMs % tickMs)      //调整当前时间      if (overflowWheel != null) overflowWheel.advanceClock(currentTime)//递归推进上级时间轮    }  }

trait Timer {  def add(timerTask: TimerTask): Unit  def advanceClock(timeoutMs: Long): Boolean  def size: Int  def shutdown(): Unit}
@threadsafeclass SystemTimer(executorName: String,                  tickMs: Long = 1,                  wheelSize: Int = 20,                  startMs: Long = Time.SYSTEM.hiResClockMs) extends Timer {
  // timeout timer  private[this] val taskExecutor = Executors.newFixedThreadPool(1,    (runnable: Runnable) => KafkaThread.nonDaemon("executor-" + executorName, runnable))  private[this] val delayQueue = new DelayQueue[TimerTaskList]()//延迟队列，会传递到每个层级的时间轮中  private[this] val taskCounter = new AtomicInteger(0)  private[this] val timingWheel = new TimingWheel(    tickMs = tickMs,    wheelSize = wheelSize,    startMs = startMs,    taskCounter = taskCounter,    delayQueue  )//...//时间轮的推进  def advanceClock(timeoutMs: Long): Boolean = {    var bucket = delayQueue.poll(timeoutMs, TimeUnit.MILLISECONDS)//从延迟队列获取槽    if (bucket != null) {      writeLock.lock()      try {        while (bucket != null) {          timingWheel.advanceClock(bucket.getExpiration())//时间轮的推进，具体看上面timingwheel中的advanceClock方法          bucket.flush(reinsert)//把槽中所有任务都取出来，并重新add一遍，此时因为任务到期，并不会真正add进去，而是调用线程池运行任务，具体看下面的reinsert和TimingWheel.add方法          bucket = delayQueue.poll()//获取下一个槽        }      } finally {        writeLock.unlock()      }      true    } else {      false    }  }private[this] val reinsert = (timerTaskEntry: TimerTaskEntry) => addTimerTaskEntry(timerTaskEntry)private def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {    if (!timingWheel.add(timerTaskEntry)) {//如果到执行时间或者取消会走到这      if (!timerTaskEntry.cancelled)//到执行时间，执行任务        taskExecutor.submit(timerTaskEntry.timerTask)    }  }//....}