package com.example.demo;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * @author zyc
 */
@Slf4j
public class ScheduledThreadPoolExecutorDemo {

    /**
     * ScheduledThreadPoolExecutor使用corePoolSize作为固定大小线程池，内部的延迟队列是一个无界队列，
     * 所以在在设置ScheduledThreadPoolExecutor的corePoolSize大小时需要注意自己有几个延迟任务需要执行，
     * 如果有4个任务的话就设置值为4，如果设置为3的话，那么很有可能第四个任务会等到其它任务执行完毕之后才能执行。
     */
    private static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(4);

    /**
     * 因为{@link ScheduledThreadPoolExecutor}会将我们提交的任务包装成一个继承{@link FutureTask}
     * 的{@code ScheduledFutureTask}，如果延迟任务执行过程中发生异常的话是不会被抛出来的，我们只能通过
     * {@link FutureTask#get()}方法来使其抛出异常，并且这个get方法会阻塞当前执行线程，为了不阻塞main线程
     * 以便调试，我们利用另一个线程池来执行每个延迟任务。
     */
    private static ExecutorService threadPoolExecutor = Executors.newFixedThreadPool(4);

    public static void main(String[] args) {
        threadPoolExecutor.execute(ScheduledThreadPoolExecutorDemo::delayRunnable);
        threadPoolExecutor.execute(ScheduledThreadPoolExecutorDemo::delayCallable);
        threadPoolExecutor.execute(ScheduledThreadPoolExecutorDemo::fixedRateRunnable);
        threadPoolExecutor.execute(ScheduledThreadPoolExecutorDemo::fixedDelayRunnable);
    }

    /**
     * 延迟2秒后执行耗时2s的Runnable任务
     */
    @SneakyThrows
    private static void delayRunnable() {
        scheduledThreadPoolExecutor.schedule(() -> {
            log.info("延迟2秒的Runnable任务开始执行");
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2));
            log.info("延迟2秒的Runnable任务执行完毕");
        }, 2, TimeUnit.SECONDS);
    }

    /**
     * 延迟2秒后执行耗时2s的Callable任务
     */
    @SneakyThrows
    private static void delayCallable() {
        scheduledThreadPoolExecutor.schedule(() -> {
            log.info("延迟2秒的Callable任务开始执行");
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2));
            log.info("延迟2秒的Callable任务执行完毕");
            return null;
        }, 2, TimeUnit.SECONDS);
    }

    /**
     * 初始延迟1秒后执行耗时4秒的Runnable任务，然后在一次任务的终止与下一次开始之间以固定的延迟2秒循环执行
     */
    @SneakyThrows
    private static void fixedDelayRunnable() {
        scheduledThreadPoolExecutor.scheduleWithFixedDelay(() -> {
            log.info("固定delay为2秒的Runnable任务开始执行");
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(4));
            log.info("固定delay为2秒的Runnable任务执行完毕");
        }, 1, 2, TimeUnit.SECONDS).get();
    }

    /**
     * 初始延迟1秒后执行耗时4秒的Runnable任务，然后每隔2秒则再次执行这个任务，需要注意的是如果2秒内这个任务没有完成的话，
     * 那么就会等待任务执行完毕后立马执行该任务，也就是说任务不一定每隔2秒就会被执行，任务执行的时机取决于任务本身的执行时长。
     */
    @SneakyThrows
    private static void fixedRateRunnable() {
        AtomicInteger i = new AtomicInteger();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            i.getAndIncrement();

            // 模拟第四次执行时发生异常
            if (i.get() == 4) {
                throw new RuntimeException();
            }
            log.info("固定Rate为2秒的Runnable任务开始执行");
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
            log.info("固定Rate为2秒的Runnable任务执行完毕");
        }, 1, 2, TimeUnit.SECONDS).get();
    }

}

package java.util.concurrent;

public interface ScheduledExecutorService extends ExecutorService {

    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);

 
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

}

public interface ScheduledFuture<V> extends Delayed, Future<V> {
}

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

public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory);
}

public ScheduledThreadPoolExecutor(int corePoolSize,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), handler);
}

public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory, handler);
}

public void execute(Runnable command) {
    schedule(command, 0, NANOSECONDS);
}

public Future<?> submit(Runnable task) {
    return schedule(task, 0, NANOSECONDS);
}

public <T> Future<T> submit(Runnable task, T result) {
    return schedule(Executors.callable(task, result), 0, NANOSECONDS);
}

public <T> Future<T> submit(Callable<T> task) {
    return schedule(task, 0, NANOSECONDS);
}

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {

    boolean isPeriodic();
}

private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {

        // 当前任务进入队列的顺序
        private final long sequenceNumber;

        // 任务将在多久后开始执行
        private long time;

         // 执行任务的周期时间。
         // 正数代表该任务是fixed-rate类型的任务
         // 负数代表任务是fixed-delay类型的任务
         // 0代表任务是一次性执行的任务
        private final long period;

       // 任务自身引用，当周期任务执行完毕后以便调用reExecutePeriodic方法重新将该任务入队
        RunnableScheduledFuture<V> outerTask = this;

        // 任务在堆中的索引
        int heapIndex;

        // 一次性任务构造器
        ScheduledFutureTask(Runnable r, V result, long ns) {
            super(r, result);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        // 周期任务构造器
        ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

       // 一次性任务构造器
        ScheduledFutureTask(Callable<V> callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

		// 计算当前任务还剩多久时间能够执行
        public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), NANOSECONDS);
        }

	   // 在任务入队时会通过这个compareTo方法来对任务执行的优先级进行比较
        public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                return 0;
             // 对于ScheduledFutureTask类型的任务首先比较的是它们的
             // 执行时间time，值越小代表越快要执行，反之则代表越晚执行。
             // 如果time相等的话则比较这两个任务入队的顺序，先入队的先执行
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
                else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            // 如果任务的类型是用户自己定义的话，则通过getDelay方法来计算任务剩余执行的时间
            // 然后进行比较，越小的值代表越先执行。
            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }

        // 判断当前任务是否是周期任务
        public boolean isPeriodic() {
            return period != 0;
        }

        // 周期任务执行完毕后调用这个方法重新设置下一次执行的时间
        private void setNextRunTime() {
            long p = period;
            // 对于fixed-ratey类型的任务来说，下次执行的时间仅仅通过
            // 上次执行的time+p，因为setNextRunTime方法是在
            // FutureTask的runAndReset方法之后被调用的（任务执行完毕后调用），
            // 如果任务执行的时间超过了period，那么这个理论上的time+p在任务被提交到
            // 队列之后的执行优先级就很高
            if (p > 0)
                time += p;
            // 对于fixed-delay类型的任务来说，每次执行的间隔都是固定好的，
            // 所以只需要将|p|+now()就是任务下一次执行的时间
            else
                time = triggerTime(-p);
        }

		// 取消任务
        public boolean cancel(boolean mayInterruptIfRunning) {
            boolean cancelled = super.cancel(mayInterruptIfRunning);
            if (cancelled && removeOnCancel && heapIndex >= 0)
                remove(this);
            return cancelled;
        }

        // 重写FutureTask的run方法逻辑
        public void run() {
            boolean periodic = isPeriodic();
            // 当前是否能够执行任务
            if (!canRunInCurrentRunState(periodic))
                cancel(false);
            // 一次性任务只需要执行一次
            else if (!periodic)
                ScheduledFutureTask.super.run();
            // 周期性任务的执行是通过FutureTask的runAndReset方法执行的
            // 执行完毕之后调用setNextRunTime方法设置下一次执行时间
            // 然后调用reExecutePeriodic方法确保只要当前线程池未关闭则必定
            // 创建一个worker线程来执行提交的任务
            else if (ScheduledFutureTask.super.runAndReset()) {
                setNextRunTime();
                reExecutePeriodic(outerTask);
            }
        }
    }

long triggerTime(long delay) {
    return now() +
        ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}