Spring任务执行器

在日常开发过程中，想要处理长时间运行的任务，对于任何应用程序开发而言都不是一件容易实现的事情。

有时候，我们需要异步执行任务或在特定延迟之后执行任务，这时候就可以通过Spring的任务执行器和任务调度器来实现这一目标。

Spring框架通过TaskExecutor和TaskScheduler接口引入了对异步执行和任务调度的抽象。

在本节中，我们先来讨论TaskExecutor。

TaskExecutor应用方式

Spring提供TaskExecutor接口作为Executor的抽象，并内置了一组非常有用的实现类。其中最具有代表性的是SimpleAsyncTaskExecutor，该类在Spring框架内部也得到了广泛应用。SimpleAsyncTaskExecutor会启动一个新线程并异步执行任务，但不会重用线程。

让我们看一个在Spring应用程序中使用SimpleAsyncTaskExecutor执行异步任务的简单示例。我们可以通过如代码清单11-5所示的方式在Spring应用程序中注入SimpleAsyncTaskExecutor。

代码清单11-5　注入SimpleAsyncTaskExecutor实现代码

@Bean

AsyncTaskExecutor taskExecutor() {

SimpleAsyncTaskExecutor taskExecutor = new

SimpleAsyncTaskExecutor();

return taskExecutor;

}

这里的AsyncTaskExecutor是TaskExecutor接口的子接口，而SimpleAsyncTaskExecutor实现了该接口。然后，我们可以构建一个AsyncTask，并通过AsyncTaskExecutor来异步执行任务，如代码清单11-6所示。

代码清单11-6　AsyncTask实现代码

public class AsyncTask {

@Autowired

private AsyncTaskExecutor executor;

public void runTasks() throws Exception {

for (int i = 1; i <= 5; i++) {

Runnable task = new SpringTask(" " + i);

executor.execute(task);

}

}

}

这里我们复用了前面所创建的SpringTask。执行这段代码，我们会在控制台中得到如代码清单11-7所示的输出结果。显然，基于SimpleAsyncTaskExecutor，任务与线程之间应该是一对一的执行关系。

代码清单11-7　Simple AsyncTaskExecutor执行日志

INFO SpringTask:15 - SimpleAsyncTaskExecutor-3, Execute Task = 3

INFO SpringTask:15 - SimpleAsyncTaskExecutor-1, Execute Task = 1

INFO SpringTask:15 - SimpleAsyncTaskExecutor-2, Execute Task = 2

INFO SpringTask:15 - SimpleAsyncTaskExecutor-5, Execute Task = 5

INFO SpringTask:15 - SimpleAsyncTaskExecutor-4, Execute Task = 4

INFO SpringTask:17 - SimpleAsyncTaskExecutor-2, End

INFO SpringTask:17 - SimpleAsyncTaskExecutor-4, End

INFO SpringTask:17 - SimpleAsyncTaskExecutor-3, EndINFO SpringTask:17 - SimpleAsyncTaskExecutor-5, End

INFO SpringTask:17 - SimpleAsyncTaskExecutor-1, End

TaskExecutor运行原理

介绍完Spring任务执行器的使用方式之后，我们来进一步分析它的实现原理。同样，我们还是以前面介绍的SimpleAsyncTaskExecutor为例展开讨论。SimpleAsyncTaskExecutor的类层结构如图11-1所示。

这里的TaskExecutor和AsyncTaskExecutor都比较好理解，我们直接看它们的接口定义，如代码清单11-8所示。

在图11-1中可以看到，AsyncListenableTaskExecutor又扩展了AsyncTaskExecutor，添加了可以返回ListenableFuture对象的方法，如代码清单11-9所示。

ListenableFuture是Spring为开发人员提供的一个工具类，它是Future的子接口，可在任务提交后添加回调，定义如代码清单11-10所示。

代码清单11-8　TaskExecutor和AsyncTaskExecutor接口定义代码

public interface TaskExecutor extends Executor {

@Override

void execute(Runnable task);

}

public interface AsyncTaskExecutor extends TaskExecutor {

void execute(Runnable task, long startTimeout);

Future<?> submit(Runnable task);

<T> Future<T> submit(Callable<T> task);

}

代码清单11-9　TaskExecutor和AsyncTaskExecutor接口定义代码

public interface AsyncListenableTaskExecutor extends AsyncTaskExecutor

{

ListenableFuture<?> submitListenable(Runnable task);

<T> ListenableFuture<T> submitListenable(Callable<T> task)

}

代码清单11-10　ListenableFuture接口定义代码

public interface ListenableFuture<T> extends Future<T> {

void addCallback(ListenableFutureCallback<? super T> callback);

void addCallback(SuccessCallback<? super T> successCallback,

FailureCallback failureCallback);

default CompletableFuture<T> completable() {

CompletableFuture<T> completable = newDelegatingCompletableFuture<>(this);

addCallback(completable::complete,

completable::completeExceptionally);

return completable;

}

}

接下来，我们来看AsyncListenableTaskExecutor中的execute()方法，如代码清单11-11所示。

代码清单11-11　AsyncListenableTaskExecutor的execute()方法代码

@Override

public void execute(Runnable task, long startTimeout) {

Assert.notNull(task, "Runnable must not be null");

//使用包装器包装任务

Runnable taskToUse = (this.taskDecorator != null ?

this.taskDecorator.decorate(task) : task);

//启用限流器来执行任务

if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {

this.concurrencyThrottle.beforeAccess();

doExecute(new ConcurrencyThrottlingRunnable(taskToUse));

}

//不启用限流器来执行任务

else {

doExecute(taskToUse);

}

}

这里引出了限流器（Throttle）的概念，限流器的作用是在线程执行的并发度达到阈值时让后续的线程处于阻塞等待。这是Spring TaskExecutor设计上的一个亮点，基本思想如图11-2所示。

通过图11-2，我们不难看出限流器在线程执行之前进行了对并发限制的

判断，如果需要限流就阻塞线程。而一旦任务执行完成，就会唤醒正在等待

的线程继续执行任务。

而真正执行任务的doExecute()方法比较简单，我们从线程工厂ThreadFactory获取线程，或者直接创建一个新的线程进行执行即可，如代码清单11-12所示。

代码清单11-12　AsyncListenableTaskExecutor的doExecute()方法代码

protected void doExecute(Runnable task) {

Thread thread = (this.threadFactory != null ?

this.threadFactory.newThread(task) : createThread(task));

thread.start();

}

本文给大家讲解的内容是并发编程：使用任务执行器和调度器，Spring任务执行器

• 下文给大家讲解的是并发编程：使用任务执行器和调度器， Spring任务调度器