jdk scheduled_thread

scheduled thread pool

构造

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE,
                DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
                new DelayedWorkQueue());
    }
}

public class ThreadPoolExecutor extends AbstractExecutorService {
    private volatile int corePoolSize;
    private final BlockingQueue<Runnable> workQueue;
    private volatile ThreadFactory threadFactory;
    private volatile RejectedExecutionHandler handler;
    private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                Executors.defaultThreadFactory(), defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        this.corePoolSize = corePoolSize;
        this.workQueue = workQueue;
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
}

ScheduledFutureTask

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
    private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
        // 当时间相等时，比较序列号
        private final long sequenceNumber;
        // 下一次执行的时间
        private volatile long time;
        // 周期
        private final long period;

        ScheduledFutureTask(Runnable r, V result, long triggerTime,
                            long sequenceNumber) {
            super(r, result);
            this.time = triggerTime;
            this.period = 0;
            this.sequenceNumber = sequenceNumber;
        }

        private void setNextRunTime() {
            long p = period;
            if (p > 0)
                time += p;
            else
                time = triggerTime(-p);
        }
    }
}

添加延迟任务

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        RunnableScheduledFuture<V> t = decorateTask(callable,
                new ScheduledFutureTask<V>(callable,
                        triggerTime(delay, unit),
                        sequencer.getAndIncrement()));
        delayedExecute(t);
        return t;
    }

    private void delayedExecute(RunnableScheduledFuture<?> task) {
        // 任务添加到队列中
        super.getQueue().add(task);
        // 忽略异常分支
        ensurePrestart();
    }
}

public class ThreadPoolExecutor extends AbstractExecutorService {
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    void ensurePrestart() {
        // 获取工作线程数量
        int wc = workerCountOf(ctl.get());
        if (wc < corePoolSize) {
            // 增加一个线程
            addWorker(null, true);
        }
    }
}

计算运行时间

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
    private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
        // 当时间相等时，比较序列号
        private final long sequenceNumber;
        // 下一次执行的时间
        private volatile long time;
        // 周期
        private final long period;

        private void setNextRunTime() {
            long p = period;
            if (p > 0)
                time += p;
            else
                time = triggerTime(-p);
        }
    }

    long triggerTime(long delay) {
        // 计算下一次触发的时间
        return System.nanoTime() +
                ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
    }
}