并发编程之定时任务&定时线程池

1. public int compareTo(Delayed other) {
2. if (other == this) // compare zero if same object
3. return 0;
4. if (other instanceof ScheduledFutureTask) {
5. ScheduledFutureTask x = (ScheduledFutureTask)other;
6. long diff = time ‐ x.time;
7. if (diff < 0)
8. return ‐1;
9. else if (diff > 0)
10. return 1;

1. else if (sequenceNumber < x.sequenceNumber)
2. return ‐1;
3. else
4. return 1;

15   }

16   long diff = getDelay(NANOSECONDS) ‐ other.getDelay(NANOSECONDS);

17   return (diff < 0) ? ‐1 : (diff > 0) ? 1 : 0;

18 }

1. public void run() {
2. boolean periodic = isPeriodic();
3. //如果当前线程池已经不支持执行任务，则取消
4. if (!canRunInCurrentRunState(periodic))
5. cancel(false);
6. //如果不需要周期性执行，则直接执行run方法然后结束
7. else if (!periodic)
8. ScheduledFutureTask.super.run();
9. //如果需要周期执行，则在执行完任务以后，设置下一次执行时间
10. else if (ScheduledFutureTask.super.runAndReset()) {
11. // 计算下次执行该任务的时间
12. setNextRunTime();
13. //重复执行任务
14. reExecutePeriodic(outerTask);

15   }

16 }

1. public ScheduledFuture schedule(Runnable command,
2. long delay,
3. TimeUnit unit) {
4. //参数校验
5. if (command == null || unit == null)
6. throw new NullPointerException();
7. //这里是一个嵌套结构，首先把用户提交的任务包装成ScheduledFutureTask
8. //然后在调用decorateTask进行包装，该方法是留给用户去扩展的，默认是个空方法
9. RunnableScheduledFuture t = decorateTask(command,
10. new ScheduledFutureTask<Void>(command, null,

1. triggerTime(delay, unit)));
2. //包装好任务以后，就进行提交了
3. delayedExecute(t);
4. return t;

15 }

1. private void delayedExecute(RunnableScheduledFuture task) {
2. //如果线程池已经关闭，则使用拒绝策略把提交任务拒绝掉
3. if (isShutdown())
4. reject(task);
5. else {
6. //与ThreadPoolExecutor不同，这里直接把任务加入延迟队列
7. super.getQueue().add(task);//使用用的DelayedWorkQueue
8. //如果当前状态无法执行任务，则取消
9. if (isShutdown() &&
10. !canRunInCurrentRunState(task.isPeriodic()) &&
11. remove(task))
12. task.cancel(false);
13. else
14. //这里是增加一个worker线程，避免提交的任务没有worker去执行
15. //原因就是该类没有像ThreadPoolExecutor一样，woker满了才放入队列
16. ensurePrestart();

17   }

18 }

1. // 队列初始容量
2. private static final int INITIAL_CAPACITY = 16;
3. // 根据初始容量创建RunnableScheduledFuture类型的数组
4. private RunnableScheduledFuture[] queue =
5. new RunnableScheduledFuture[INITIAL_CAPACITY];
6. private final ReentrantLock lock = new ReentrantLock();
7. private int size = 0;
8. // leader线程
9. private Thread leader = null;
10. // 当较新的任务在队列的头部可用时，或者新线程可能需要成为leader，则通过该条件发出信号
11. private final Condition available = lock.newCondition();

1. public boolean offer(Runnable x) {
2. //参数校验
3. if (x == null)
4. throw new NullPointerException();
5. RunnableScheduledFuture e = (RunnableScheduledFuture)x;
6. final ReentrantLock lock = this.lock;
7. lock.lock();
8. try {
9. //查看当前元素数量，如果大于队列长度则进行扩容

1. public RunnableScheduledFuture take() throws InterruptedExcep tion {
2. final ReentrantLock lock = this.lock;
3. lock.lockInterruptibly();
4. try {

5   for (;;) {

1. RunnableScheduledFuture first = queue[0];
2. if (first == null)
3. available.await();
4. else {
5. // 计算当前时间到执行时间的时间间隔
6. long delay = first.getDelay(NANOSECONDS);
7. if (delay <= 0)
8. return finishPoll(first);
9. first = null; // don't retain ref while waiting

1. // leader不为空，阻塞线程
2. if (leader != null)
3. available.await();
4. else {
5. // leader为空，则把leader设置为当前线程，
6. Thread thisThread = Thread.currentThread();
7. leader = thisThread;
8. try {
9. // 阻塞到执行时间
10. available.awaitNanos(delay);
11. } finally {
12. // 设置leader = null，让其他线程执行available.awaitNanos(delay);
13. if (leader == thisThread)
14. leader = null;

29   }

30   }

31   }

32   }

1. } finally {
2. // 如果leader不为空，则说明leader的线程正在执行available.awaitNano s(delay);
3. // 如果queue[0] == null，说明队列为空
4. if (leader == null && queue[0] != null)
5. available.signal();
6. lock.unlock();

39   }

40 }

1. public RunnableScheduledFuture poll(long timeout, TimeUnit un it)
2. throws InterruptedException {
3. long nanos = unit.toNanos(timeout);
4. final ReentrantLock lock = this.lock;
5. lock.lockInterruptibly();
6. try {

7   for (;;) {

1. RunnableScheduledFuture first = queue[0];
2. if (first == null) {
3. if (nanos <= 0)
4. return null;
5. else
6. nanos = available.awaitNanos(nanos);
7. } else {
8. long delay = first.getDelay(NANOSECONDS);
9. // 如果delay <= 0，说明已经到了任务执行的时间，返回。
10. if (delay <= 0)
11. return finishPoll(first);
12. // 如果nanos <= 0，说明已经超时，返回null
13. if (nanos <= 0)
14. return null;
15. first = null; // don't retain ref while waiting
16. // nanos < delay 说明需要等待的时间小于任务要执行的延迟时间

1. // leader != null 说明有其它线程正在对任务进行阻塞
2. // 这时阻塞当前线程nanos纳秒
3. if (nanos < delay || leader != null)
4. nanos = available.awaitNanos(nanos);
5. else {
6. Thread thisThread = Thread.currentThread();
7. leader = thisThread;
8. try {
9. // 这里的timeLeft表示delay减去实际的等待时间
10. long timeLeft = available.awaitNanos(delay);
11. // 计算剩余的等待时间
12. nanos ‐= delay ‐ timeLeft;
13. } finally {
14. if (leader == thisThread)
15. leader = null;

39   }

40   }

41   }

42   }

1. } finally {
2. if (leader == null && queue[0] != null)
3. available.signal();
4. lock.unlock();

47   }

48 }

1. private RunnableScheduledFuture finishPoll(RunnableScheduledF uture f) {
2. // 数组长度‐1
3. int s = ‐‐size;
4. // 取出最后一个节点
5. RunnableScheduledFuture x = queue[s];
6. queue[s] = null;

7 // 长度不为0，则从第一个元素开始排序，目的是要把最后一个节点放到合适的位置上

8   if (s != 0)

1. siftDown(0, x);
2. setIndex(f, ‐1);
3. return f;

12 }

1. private void siftDown(int k, RunnableScheduledFuture key) {
2. // 根据二叉树的特性，数组长度除以2，表示取有子节点的索引
3. int half = size >>> 1;
4. // 判断索引为k的节点是否有子节点
5. while (k < half) {
6. // 左子节点的索引
7. int child = (k << 1) + 1;
8. RunnableScheduledFuture c = queue[child];
9. // 右子节点的索引
10. int right = child + 1;
11. // 如果有右子节点并且左子节点的时间间隔大于右子节点，取时间间隔最小的节点
12. if (right < size && c.compareTo(queue[right]) > 0)
13. c = queue[child = right];
14. // 如果key的时间间隔小于等于c的时间间隔，跳出循环
15. if (key.compareTo(c) <= 0)
16. break;
17. // 设置要移除索引的节点为其子节点
18. queue[k] = c;
19. setIndex(c, k);
20. k = child;

21   }

1. // 将key放入索引为k的位置
2. queue[k] = key;
3. setIndex(key, k);

25 }

1. public boolean remove(Object x) {
2. final ReentrantLock lock = this.lock;
3. lock.lock();
4. try {
5. int i = indexOf(x);

6   if (i < 0)

7   return false;

8

1. setIndex(queue[i], ‐1);
2. int s = ‐‐size;
3. RunnableScheduledFuture replacement = queue[s];
4. queue[s] = null;

13   if (s != i) {

14   // 从i开始向下调整

1. siftDown(i, replacement);
2. // 如果queue[i] == replacement，说明i是叶子节点
3. // 如果是这种情况，不能保证子节点的下次执行时间比父节点的大
4. // 这时需要进行一次向上调整
5. if (queue[i] == replacement)
6. siftUp(i, replacement);

21   }

1. return true;
2. } finally {
3. lock.unlock();

25   }

26 }