问题原因

在公司写项目的时候，有一个周期定时任务的需求，就想着阿里巴巴开发手册里不是说不能用Executors去创建线程池，因为存在如下问题：

• FixedThreadPool和SingleThreadPool：允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
• CachedThreadPool：允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

然后就没用Executors.newScheduledThreadPool()，然后自己new一个ScheduledThreadPoolExecutor对象，并重写了afterExecute方法，和自定义拒绝策略。

结果运行起来只执行一次就不打印日志了，这个问题困扰了我半天，所以留个笔记记录下来。
代码如下：

@Slf4j
@Component
public class PlanStartAndEndTask implements ApplicationRunner {

  /**
   * 初始化定时任务线程池
   */
  private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, new RecordExceptionExecutionHandler()) {

    /**
     * 自定义异常处理
     * @param runnable 任务
     * @param throwable 异常
     * @date 2021/3/31
     */
    @Override
    protected void afterExecute(Runnable runnable, Throwable throwable) {
      final Logger log = LoggerFactory.getLogger(this.getClass());
      if (runnable instanceof Thread) {
        if (throwable != null) {
          log.error("自动开始/结束分享计划的定时任务出现异常,时间：{},异常信息：{}", LocalDateTime.now(), throwable.getMessage());
        }
      } else if (runnable instanceof FutureTask) {
        FutureTask<?> futureTask = (FutureTask<?>) runnable;
        try {
          // 问题就出在这！！！
          futureTask.get();
        } catch (InterruptedException e) {
          log.error("自动开始/结束分享计划的定时任务被打断,时间：{}", LocalDateTime.now());
          Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
          log.error("自动开始/结束分享计划的定时任务出现异常,时间：{},异常信息：{}", LocalDateTime.now(), e.getMessage());
        }
      }
    }
  };

  @Override
  public void run(ApplicationArguments args) throws Exception {
    // 为了模拟，首次延时时间0，周期为5秒钟一次
    executor.scheduleAtFixedRate(() -> {
      long startTime = System.currentTimeMillis();
        log.info("开始执行自动化任务");
        /** 省略业务代码 **/
        log.info("结束执行自动化任务,耗时：{}毫秒;", System.currentTimeMillis() - startTime);
    }, 0, 5000, TimeUnit.MILLISECONDS);
  }

  /**
   * 自定义实现拒绝策略，记录日志，队列满了之后，新任务被提交会直接被丢弃掉
   *
   * @author Zhu Lin
   * @date 2021/3/30
   */
  @Slf4j
  static class RecordExceptionExecutionHandler implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution(Runnable runnable, ThreadPoolExecutor threadPoolExecutor) {
      log.error("任务：{}, 被{}拒绝 ", runnable.toString(), threadPoolExecutor.toString());
    }
  }

}

然后起初我以为是异常的问题导致只执行一次就不执行了，控制台也不打印异常信息，因为JavaDoc中也是这么说的

    /**
     * Creates and executes a periodic action that becomes enabled first
     * after the given initial delay, and subsequently with the given
     * period; that is executions will commence after
     * {@code initialDelay} then {@code initialDelay+period}, then
     * {@code initialDelay + 2 * period}, and so on.
     * If any execution of the task
     * encounters an exception, subsequent executions are suppressed.
     * Otherwise, the task will only terminate via cancellation or
     * termination of the executor.  If any execution of this task
     * takes longer than its period, then subsequent executions
     * may start late, but will not concurrently execute.
     *
     * @param command the task to execute
     * @param initialDelay the time to delay first execution
     * @param period the period between successive executions
     * @param unit the time unit of the initialDelay and period parameters
     * @return a ScheduledFuture representing pending completion of
     *         the task, and whose {@code get()} method will throw an
     *         exception upon cancellation
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if command is null
     * @throws IllegalArgumentException if period less than or equal to zero
     */
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

简单来说就是，任务只有遇到异常时才会停止，否贼只有取消和终止执行程序才能终止任务
所以我就改成这样

  @Override
  public void run(ApplicationArguments args) throws Exception {
    // 首次延时时间0，周期为5秒钟一次
    executor.scheduleAtFixedRate(() -> {
      long startTime = System.currentTimeMillis();
      try {
        log.info("开始执行自动化任务");
      } catch (Exception e) {
        log.error("自动开始/结束分享计划的定时任务出现异常,时间：{},异常信息：{}", LocalDateTime.now(), e.getMessage());
      } finally {
        log.info("结束执行自动化任务,耗时：{}毫秒;", System.currentTimeMillis() - startTime);
      }
    }, 0, 5000, TimeUnit.MILLISECONDS);
  }

事实证明压根不是这个问题，毕竟我里面啥都没干，就打印日志，抛啥子异常咯~，然后我试了好多遍，终于在我把afterExecutor给注释掉后，程序居然正常了！最后我把问题定位到 futureTask.get() 这行代码上，通过debug发现，线程执行到这行代码之后就不会往下走了，那么原因到底是什么？我们深入来看一下

问题解析

首先我们先看看为什么get()会被阻塞住

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 如果任务已经在执行中了，那么就进入等待
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

// 等待任务执行完成
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    // 计算等待终止时间，如果一直等待的话，终止时间为 0
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    // 不排队
    boolean queued = false;
    // 无限循环
    for (;;) {
        // 如果线程已经被打断了，删除，抛异常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        // 当前任务状态
        int s = state;
        // 当前任务已经执行完了，返回
        if (s > COMPLETING) {
            // 当前任务的线程置空
            if (q != null)
                q.thread = null;
            return s;
        }
        // 如果正在执行，当前线程让出 cpu，重新竞争，防止 cpu 飙高
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
            // 如果第一次运行，新建 waitNode，当前线程就是 waitNode 的属性
        else if (q == null)
            q = new WaitNode();
            // 默认第一次都会执行这里，执行成功之后，queued 就为 true，就不会再执行了
            // 把当前 waitNode 当做 waiters 链表的第一个
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
            // 如果设置了超时时间，并过了超时时间的话，从 waiters 链表中删除当前 wait
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            // 没有过超时时间，线程进入 TIMED_WAITING 状态
            LockSupport.parkNanos(this, nanos);
        }
        // 没有设置超时时间，进入 WAITING 状态
        else
            LockSupport.park(this);
    }
}

我们可以看到上面那行注释为“当前任务已经执行完了，返回”的代码，只要不满足这个条件，你就会被一直阻塞，那么问题肯定出在我提交的定时任务state从来就没有被改变，这又是为什么？我们继续深究
接下来我们看到ScheduledThreadPoolExecutor#scheduleAtFixedRate方法

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(period));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

这里的核心逻辑就是将 Runnable 包装成了一个ScheduledFutureTask对象，这个包装是在FutureTask基础上增加了定时调度需要的一些数据。（FutureTask是线程池的核心类之一）decorateTask是一个钩子方法，用来给扩展用的，在这里的默认实现就是返回ScheduledFutureTask本身。
然后主逻辑就是通过delayedExecute放入队列中。

那么为什么我们的任务state状态没有改变，肯定就是ScheduledFutureTask的run方法啦。

/**
 * Overrides FutureTask version so as to reset/requeue if periodic.
 */

public void run() {
    // 是否是周期性任务
    boolean periodic = isPeriodic();
    // 如果不可以在当前状态下运行，就取消任务（将这个任务的状态设置为CANCELLED）。
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    else if (!periodic)
        // 如果不是周期性的任务，调用 FutureTask # run 方法
        ScheduledFutureTask.super.run();
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 如果是周期性的，设置下次执行时间
        setNextRunTime();
        // 再次将任务添加到队列中
        reExecutePeriodic(outerTask);
    }
}

我们重点看ScheduledFutureTask.super.runAndReset()方法，实际上调用的是其父类FutureTask的runAndReset()方法，这个方法会在执行成功之后重置线程状态，reset就是这个语义。同时我们可以看到，当方法执行返回false的时候，就不会再次将任务添加的队列中，这和我们最开始假设的异常情况是一致的
最后答案就在这个runAndReset和run方法的区别里：

    public void run() {
        /** 省略其他代码 **/
        try {
            // 执行任务
            result = c.call();
            ran = true;
        } catch (Throwable ex) {
            result = null;
            ran = false;
            setException(ex);
        }
        if (ran)
            set(result);
        /** 省略其他代码 **/
    }

    protected boolean runAndReset() {
        /** 省略其他代码 **/
        try {
            // 执行任务
            c.call(); // don't set result
            ran = true;
        } catch (Throwable ex) {
            setException(ex);
        }
        /** 省略其他代码 **/
    }

上面的代码我省略掉了大部分，只留出了这次问题所在的地方，感兴趣的小伙伴可以自己去ide里看看，c.call()是执行任务的地方，这里有一个默认为false的ran变量，当任务执行成功时，ran会被设成 true，即任务已执行。但这不是关键，关键是我们发现run方法里在成功后回去调一个set方法

    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

在set方法中修改了state的状态，这也证明了我们之前的逻辑，周期任务调runAndReset压根不去修改state，所以get方法只能阻塞，没有其他选择。

小疑惑

其实如果按阿里巴巴开发手册的规范来说的话，ScheduledThreadPoolExecutor也存在允许创建的线程数据为Integer.MAX_VALUE的问题，那么该怎么解决呢，这点我有点疑惑。

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