Spring中异步和计划任务如何实现

这篇文章将为大家详细讲解有关Spring中异步和计划任务如何实现，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。

专注于为中小企业提供网站建设、做网站服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业芮城免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了上千企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。

什么是Spring中的异步任务？

在我们正式的进入话题之前，对于Spring，我们需要理解下它实现的两个不同的概念：异步任务和调度任务。显然，两者有一个很大的共同点：都在后台工作。但是，它们之间存在了很大差异。调度任务与异步不同，其作用与Linux中的CRON job完全相同(windows里面也有计划任务)。举个栗子，有一个任务必须每40分钟执行一次，那么，可以通过XML文件或者注解来进行此配置。简单的异步任务在后台执行就好，无需配置执行频率。

因为它们是两种不同的任务类型，它们两个的执行者自然也就不同。第一个看起来有点像Java的并发执行器(concurrency executor)，这里会有专门去写一篇关于Java中的执行器来具体了解。根据Spring文档TaskExecutor所述，它提供了基于Spring的抽象来处理线程池，这点，也可以通过其类的注释去了解。另一个抽象接口是TaskScheduler，它用于在将来给定的时间点来安排任务，并执行一次或定期执行。

在分析源码的过程中，发现另一个比较有趣的点是触发器。它存在两种类型：CronTrigger或PeriodTrigger。第一个模拟CRON任务的行为。所以我们可以在将来确切时间点提交一个任务的执行。另一个触发器可用于定期执行任务。

Spring的异步任务类

让我们从org.springframework.core.task.TaskExecutor类的分析开始。你会发现，其简单的不行，它是一个扩展Java的Executor接口的接口。它的唯一方法也就是是执行，在参数中使用Runnable类型的任务。

package org.springframework.core.task;
import java.util.concurrent.Executor;
/**
 * Simple task executor interface that abstracts the execution
 * of a {@link Runnable}.
 *
 * 
Implementations can use all sorts of different execution strategies,
 * such as: synchronous, asynchronous, using a thread pool, and more.
 *
 * 
Equivalent to JDK 1.5's {@link java.util.concurrent.Executor}
 * interface; extending it now in Spring 3.0, so that clients may declare
 * a dependency on an Executor and receive any TaskExecutor implementation.
 * This interface remains separate from the standard Executor interface
 * mainly for backwards compatibility with JDK 1.4 in Spring 2.x.
 *
 * @author Juergen Hoeller
 * @since 2.0
 * @see java.util.concurrent.Executor
 */
@FunctionalInterface
public interface TaskExecutor extends Executor {
 /**
 * Execute the given {@code task}.
 * 
The call might return immediately if the implementation uses
 * an asynchronous execution strategy, or might block in the case
 * of synchronous execution.
 * @param task the {@code Runnable} to execute (never {@code null})
 * @throws TaskRejectedException if the given task was not accepted
 */
 @Override
 void execute(Runnable task);
}

相对来说，org.springframework.scheduling.TaskScheduler接口就有点复杂了。它定义了一组以schedule开头的名称的方法允许我们定义将来要执行的任务。所有 schedule* 方法返回java.util.concurrent.ScheduledFuture实例。Spring5中对scheduleAtFixedRate方法做了进一步的充实，其实最终调用的还是ScheduledFuture scheduleAtFixedRate(Runnable task, long period);

public interface TaskScheduler {
 /**
 * Schedule the given {@link Runnable}, invoking it whenever the trigger
 * indicates a next execution time.
 * 
Execution will end once the scheduler shuts down or the returned
 * {@link ScheduledFuture} gets cancelled.
 * @param task the Runnable to execute whenever the trigger fires
 * @param trigger an implementation of the {@link Trigger} interface,
 * e.g. a {@link org.springframework.scheduling.support.CronTrigger} object
 * wrapping a cron expression
 * @return a {@link ScheduledFuture} representing pending completion of the task,
 * or {@code null} if the given Trigger object never fires (i.e. returns
 * {@code null} from {@link Trigger#nextExecutionTime})
 * @throws org.springframework.core.task.TaskRejectedException if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 * @see org.springframework.scheduling.support.CronTrigger
 */
 @Nullable
 ScheduledFuture schedule(Runnable task, Trigger trigger);
 
 /**
 * Schedule the given {@link Runnable}, invoking it at the specified execution time.
 * 
Execution will end once the scheduler shuts down or the returned
 * {@link ScheduledFuture} gets cancelled.
 * @param task the Runnable to execute whenever the trigger fires
 * @param startTime the desired execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @return a {@link ScheduledFuture} representing pending completion of the task
 * @throws org.springframework.core.task.TaskRejectedException if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 * 使用了默认实现，值得我们学习使用的，Java9中同样可以有私有实现的，从这里我们可以做到我只通过  * 一个接口你来实现，我把其他相应的功能默认实现下，最后调用你自定义实现的接口即可，使接口功能更  * 加一目了然
 * @since 5.0
 * @see #schedule(Runnable, Date)
 */
 default ScheduledFuture schedule(Runnable task, Instant startTime) {
 return schedule(task, Date.from(startTime));
 }
 /**
 * Schedule the given {@link Runnable}, invoking it at the specified execution time.
 * 
Execution will end once the scheduler shuts down or the returned
 * {@link ScheduledFuture} gets cancelled.
 * @param task the Runnable to execute whenever the trigger fires
 * @param startTime the desired execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @return a {@link ScheduledFuture} representing pending completion of the task
 * @throws org.springframework.core.task.TaskRejectedException if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 */
 ScheduledFuture schedule(Runnable task, Date startTime);
...
/**
 * Schedule the given {@link Runnable}, invoking it at the specified execution time
 * and subsequently with the given period.
 * 
Execution will end once the scheduler shuts down or the returned
 * {@link ScheduledFuture} gets cancelled.
 * @param task the Runnable to execute whenever the trigger fires
 * @param startTime the desired first execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @param period the interval between successive executions of the task
 * @return a {@link ScheduledFuture} representing pending completion of the task
 * @throws org.springframework.core.task.TaskRejectedException if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 * @since 5.0
 * @see #scheduleAtFixedRate(Runnable, Date, long)
 */
 default ScheduledFuture scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
 return scheduleAtFixedRate(task, Date.from(startTime), period.toMillis());
 }
 /**
 * Schedule the given {@link Runnable}, invoking it at the specified execution time
 * and subsequently with the given period.
 * 
Execution will end once the scheduler shuts down or the returned
 * {@link ScheduledFuture} gets cancelled.
 * @param task the Runnable to execute whenever the trigger fires
 * @param startTime the desired first execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @param period the interval between successive executions of the task (in milliseconds)
 * @return a {@link ScheduledFuture} representing pending completion of the task
 * @throws org.springframework.core.task.TaskRejectedException if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 */
 ScheduledFuture scheduleAtFixedRate(Runnable task, Date startTime, long period);
...
}

之前提到两个触发器组件，都实现了org.springframework.scheduling.Trigger接口。这里，我们只需关注一个的方法nextExecutionTime ，其定义下一个触发任务的执行时间。它的两个实现，CronTrigger和PeriodicTrigger，由org.springframework.scheduling.TriggerContext来实现信息的存储，由此，我们可以很轻松获得一个任务的最后一个执行时间(lastScheduledExecutionTime)，给定任务的最后完成时间(lastCompletionTime)或最后一个实际执行时间(lastActualExecutionTime)。接下来，我们通过阅读源代码来简单的了解下这些东西。org.springframework.scheduling.concurrent.ConcurrentTaskScheduler包含一个私有类EnterpriseConcurrentTriggerScheduler。在这个class里面，我们可以找到schedule方法：

public ScheduledFuture schedule(Runnable task, final Trigger trigger) {
 ManagedScheduledExecutorService executor = (ManagedScheduledExecutorService) scheduledExecutor;
 return executor.schedule(task, new javax.enterprise.concurrent.Trigger() {
  @Override
  public Date getNextRunTime(LastExecution le, Date taskScheduledTime) {
   return trigger.nextExecutionTime(le != null ?
    new SimpleTriggerContext(le.getScheduledStart(), le.getRunStart(), le.getRunEnd()) :
    new SimpleTriggerContext());
  }
  @Override
  public boolean skipRun(LastExecution lastExecution, Date scheduledRunTime) {
   return false;
  }
 });
}

SimpleTriggerContext从其名字就可以看到很多东西了，因为它实现了TriggerContext接口。

/**
 * Simple data holder implementation of the {@link TriggerContext} interface.
 *
 * @author Juergen Hoeller
 * @since 3.0
 */
public class SimpleTriggerContext implements TriggerContext {
 @Nullable
 private volatile Date lastScheduledExecutionTime;
 @Nullable
 private volatile Date lastActualExecutionTime;
 @Nullable
 private volatile Date lastCompletionTime;
...
 /**
 * Create a SimpleTriggerContext with the given time values.
 * @param lastScheduledExecutionTime last scheduled execution time
 * @param lastActualExecutionTime last actual execution time
 * @param lastCompletionTime last completion time
 */
 public SimpleTriggerContext(Date lastScheduledExecutionTime, Date lastActualExecutionTime, Date lastCompletionTime) {
 this.lastScheduledExecutionTime = lastScheduledExecutionTime;
 this.lastActualExecutionTime = lastActualExecutionTime;
 this.lastCompletionTime = lastCompletionTime;
 }
 ...
}

也正如你看到的，在构造函数中设置的时间值来自javax.enterprise.concurrent.LastExecution的实现，其中：

• getScheduledStart：返回上次开始执行任务的时间。它对应于TriggerContext的lastScheduledExecutionTime属性。

• getRunStart：返回给定任务开始运行的时间。在TriggerContext中，它对应于lastActualExecutionTime。

• getRunEnd：任务终止时返回。它用于在TriggerContext中设置lastCompletionTime。

Spring调度和异步执行中的另一个重要类是org.springframework.core.task.support.TaskExecutorAdapter。它是一个将java.util.concurrent.Executor作为Spring基本的执行器的适配器(描述的有点拗口,看下面代码就明了了)，之前已经描述了TaskExecutor。实际上，它引用了Java的ExecutorService，它也是继承了Executor接口。此引用用于完成所有提交的任务。

/**
 * Adapter that takes a JDK {@code java.util.concurrent.Executor} and
 * exposes a Spring {@link org.springframework.core.task.TaskExecutor} for it.
 * Also detects an extended {@code java.util.concurrent.ExecutorService 从此解释上面的说明}, adapting
 * the {@link org.springframework.core.task.AsyncTaskExecutor} interface accordingly.
 *
 * @author Juergen Hoeller
 * @since 3.0
 * @see java.util.concurrent.Executor
 * @see java.util.concurrent.ExecutorService 
 * @see java.util.concurrent.Executors
 */
public class TaskExecutorAdapter implements AsyncListenableTaskExecutor {
 private final Executor concurrentExecutor;
 @Nullable
 private TaskDecorator taskDecorator;
 ...
  /**
 * Create a new TaskExecutorAdapter,
 * using the given JDK concurrent executor.
 * @param concurrentExecutor the JDK concurrent executor to delegate to
 */
 public TaskExecutorAdapter(Executor concurrentExecutor) {
 Assert.notNull(concurrentExecutor, "Executor must not be null");
 this.concurrentExecutor = concurrentExecutor;
 }
  /**
 * Delegates to the specified JDK concurrent executor.
 * @see java.util.concurrent.Executor#execute(Runnable)
 */
 @Override
 public void execute(Runnable task) {
 try {
  doExecute(this.concurrentExecutor, this.taskDecorator, task);
 }
 catch (RejectedExecutionException ex) {
  throw new TaskRejectedException(
   "Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
 }
 }
 @Override
 public void execute(Runnable task, long startTimeout) {
 execute(task);
 }
 @Override
 public Future submit(Runnable task) {
 try {
  if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
  return ((ExecutorService) this.concurrentExecutor).submit(task);
  }
  else {
  FutureTask