Spring5源码解析之Spring中的异步和计划任务

发布时间：2020-05-23 12:31:09 所属栏目：Java 来源：互联网

导读：Java提供了许多创建线程池的方式，并得到一个Future实例来作为任务结果。对于Spring同样小菜一碟，通过其scheduling包就可以做到将任务线程中后台执行。

Java提供了许多创建线程池的方式，并得到一个Future实例来作为任务结果。对于Spring同样小菜一碟，通过其scheduling包就可以做到将任务线程中后台执行。

在本文的第一部分中，我们将讨论下Spring中执行计划任务的一些基础知识。之后，我们将解释这些类是如何一起协作来启动并执行计划任务的。下一部分将介绍计划和异步任务的配置。最后，我们来写个Demo，看看如何通过单元测试来编排计划任务。

什么是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}.
 *
 * <p>Implementations can use all sorts of different execution strategies,* such as: synchronous,asynchronous,using a thread pool,and more.
 *
 * <p>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}.
 * <p>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.
 * <p>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.
 * <p>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},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.
 * <p>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,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},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;
  }
 });
}

（编辑：安卓应用网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!