Timer源码解读

Timer

Timer在JDK5.0之前是唯一的内置任务调度方法。其惯常的使用方式是

final Timer timer = new Timer();
timer.schedule(new TimerTask() {
     @Override
     public void run() {
          // do sth
     }, 10*1000, 10*1000);

我们来看一下Timer的源码是怎样的。按照其使用方式,可以猜到:

  • Timer内部包含一个任务队列
  • 任务队列的元素应该是TimerTask

首先看一下TimerTask

TimerTask

public abstract class TimerTask implements Runnable {
    /**
     * This object is used to control access to the TimerTask internals.
     */
    final Object lock = new Object();

    /**
     * The state of this task, chosen from the constants below.
     */
    int state = VIRGIN;
    
    /**
     * Next execution time for this task in the format returned by
     * System.currentTimeMillis, assuming this task is scheduled for execution.
     * For repeating tasks, this field is updated prior to each task execution.
     */
    long nextExecutionTime;

    /**
     * Period in milliseconds for repeating tasks.  A positive value indicates
     * fixed-rate execution.  A negative value indicates fixed-delay execution.
     * A value of 0 indicates a non-repeating task.
     */
    long period = 0;

从源代码中可以看到TimerTask只是在Runnable上只是做了一个非常简单的扩展,增加了4个属性字段,分别记录一些属性值。

在Timer中有多个不同的构造方法,最终多个不同的构造方法会最终都是重载使用了下面这个方法

private void sched(TimerTask task, long time, long period) {
        if (time < 0)
            throw new IllegalArgumentException("Illegal execution time.");

        // Constrain value of period sufficiently to prevent numeric
        // overflow while still being effectively infinitely large.
        if (Math.abs(period) > (Long.MAX_VALUE >> 1))
            period >>= 1;

        synchronized(queue) {
            if (!thread.newTasksMayBeScheduled)
                throw new IllegalStateException("Timer already cancelled.");

            synchronized(task.lock) {
                if (task.state != TimerTask.VIRGIN)
                    throw new IllegalStateException(
                        "Task already scheduled or cancelled");
                task.nextExecutionTime = time;
                task.period = period;
                task.state = TimerTask.SCHEDULED;
            }

            queue.add(task);
            if (queue.getMin() == task)
                queue.notify();
        }
    }

继续看源码,添加任务时,可以看到任务最终被插入到一个队列中了。而在在Timer代码的一开头,我们就看到这里确实有一个队列

public class Timer {
    /**
     * The timer task queue.  This data structure is shared with the timer
     * thread.  The timer produces tasks, via its various schedule calls,
     * and the timer thread consumes, executing timer tasks as appropriate,
     * and removing them from the queue when they're obsolete.
     */
    private final TaskQueue queue = new TaskQueue();

    /**
     * The timer thread.
     */
    private final TimerThread thread = new TimerThread(queue);

TaskQueue内部存储着所有待执行的TimerTask,而TimerThread则是所有TimerTask的执行线程

TaskQueue

TaskQueue内部包含一个128位的TimerTask数组,所有的TimerTask在数组内以小顶堆的方式进行存储。

class TaskQueue {
    /**
     * Priority queue represented as a balanced binary heap: the two children
     * of queue[n] are queue[2*n] and queue[2*n+1].  The priority queue is
     * ordered on the nextExecutionTime field: The TimerTask with the lowest
     * nextExecutionTime is in queue[1] (assuming the queue is nonempty).  For
     * each node n in the heap, and each descendant of n, d,
     * n.nextExecutionTime <= d.nextExecutionTime.
     */
    private TimerTask[] queue = new TimerTask[128];

上面的代码里,堆首的位置是1,而不是0;位置是1的时候,堆中parent和两个child的位置运行关系恰好是[n]->[2n],[2n+1],非常便于理解和运算。如果使用0位置作为堆首,那么运算关系则需要变成:[n]->[2n+1],[2n+2],这个运算关系看起来比较奇怪。

private void fixUp(int k) {
        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;
        }
    }
    
    private void fixDown(int k) {
        int j;
        while ((j = k << 1) <= size && j > 0) {
            if (j < size &&
                queue[j].nextExecutionTime > queue[j+1].nextExecutionTime)
                j++; // j indexes smallest kid
            if (queue[k].nextExecutionTime <= queue[j].nextExecutionTime)
                break;
            TimerTask tmp = queue[j];  queue[j] = queue[k]; queue[k] = tmp;
            k = j;
        }
    }

    void heapify() {
        for (int i = size/2; i >= 1; i--)
            fixDown(i);
    }

很标准的堆操作

TimerThread

TimerThread extend thread,它是Timer中所有任务实际的执行线程,thread的run方法中只有一个mainloop方法,其中执行了一个无限循环。循环中,从队列中拿到最近需要执行的一个任务,

public void run() {
        try {
            mainLoop();
        } finally {
            // Someone killed this Thread, behave as if Timer cancelled
            synchronized(queue) {
                newTasksMayBeScheduled = false;
                queue.clear();  // Eliminate obsolete references
            }
        }
    }

    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) { // Non-repeating, remove
                                queue.removeMin();
                                task.state = TimerTask.EXECUTED;
                            } else { // Repeating task, reschedule
                                queue.rescheduleMin(
                                  task.period<0 ? currentTime   - task.period
                                                : executionTime + task.period);
                            }
                        }
                    }
                    if (!taskFired) // Task hasn't yet fired; wait
                        queue.wait(executionTime - currentTime);
                }
                if (taskFired)  // Task fired; run it, holding no locks
                    task.run();
            } catch(InterruptedException e) {
            }
        }
    }

当queue里面为空时,则wait等待(当向queue中add时,此处的wait则会被唤醒),唤醒之后则获取queue[1]位置的task,确定该Task是否到了执行时间,如果到了执行时间,则开始执行。

上面的代码里面可以关注到几个细节:

  • Timer的任务的并不能保证严格准时的任务执行,它的执行时机是这一次CPU调度到这个线程时,有一个任务的执行时间刚好到了或者已经过了
  • 所有的Task都是在这一个线程内执行的,只有执行完一个Task的run方法,才回去取下一个任务执行。因此加入Timer内加入的Task执行时间过长,超过了下一个任务的开始时间,那么下一个任务的执行将会延迟。
  • 在mainloop中如果发生的是中断异常,单次任务会停止;如果发生的是任何其他异常,整个mainloop将完全停止,队列清空后线程停止

上面的几点也刚好是Timer本身的主要缺陷。所以后续Java推出了java.util.concurrent.ScheduledExecutorService弥补了这些缺陷。

  • Previous

    Git

原文 

https://chchaooo.github.io/2018/12/27/Timer源码解读/

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » Timer源码解读

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址