SpringBoot异步调用--@Async详解

1. 概述

  在日常开发中，为了提高主线程的效率，往往需要采用异步调用处理，例如系统日志等。在实际业务场景中，可以使用消息中间件如RabbitMQ、RocketMQ、Kafka等来解决。假如对高可用没有太高的要求，也可以使用线程池或者队列来解决。

2. 创建工程

• 创建Maven工程
  
• 修改配置文件

<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.c3stones</groupId>
	<artifactId>spring-boot-async-demo</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>spring-boot-async-demo</name>
	<description>Spring Boot Simple Demo</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.2.6.RELEASE</version>
		<relativePath />
	</parent>

	<properties>
		<java.version>1.8</java.version>
		<maven-jar-plugin.version>3.1.1</maven-jar-plugin.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>com.google.guava</groupId>
			<artifactId>guava</artifactId>
			<version>28.2-jre</version>
		</dependency>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>

</project>

• 示例1：创建线程池，异步调用

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

public class SimpleDemo {

	// 创建固定数目线程的线程池
	// 阿里巴巴Java开发手册中提到可能存在OOM（OutOfMemory）内存溢出异常
//	private static ExecutorService executorService = Executors.newFixedThreadPool(5);

	// 推荐使用com.google.guava的ThreadFactoryBuilder来创建线程池
	private static ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("simple-threadpool-%d")
			.build();
	/**
	 * java.util.concurrent.ThreadPoolExecutor.ThreadPoolExecutor(int corePoolSize,
	 * int maximumPoolSize, long keepAliveTime, TimeUnit unit,
	 * BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
	 * RejectedExecutionHandler handler)
	 * 
	 * @param corePoolSize    线程池大小
	 * @param maximumPoolSize 最大线程数
	 * @param keepAliveTime   当线程大于线程池大小时，最长等待时间
	 * @param unit            时间单位
	 * @param workQueue       任务队列
	 * @param threadFactory   指定线程工厂
	 * @param handler         当线程池界限和队列容量时，阻止线程处理
	 */
	private static ExecutorService threadPool = new ThreadPoolExecutor(5, 200, 0L, TimeUnit.MILLISECONDS,
			new LinkedBlockingQueue<Runnable>(1024), threadFactory, new ThreadPoolExecutor.AbortPolicy());

	public static void main(String[] args) {
		// 提交线程到线程池
		for (int i = 0; i < 5; i++) {
			threadPool.execute(new SimpleThread1());
			threadPool.execute(new SimpleThread2());
		}

		// 关闭
		threadPool.shutdown();
	}
}

class SimpleThread1 implements Runnable {

	private static Logger logger = LoggerFactory.getLogger(SimpleThread1.class);

	@Override
	public void run() {
		try {
			logger.info("线程 SimpleThread1 被调用！");
			Thread.sleep(3000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

class SimpleThread2 implements Runnable {

	private static Logger logger = LoggerFactory.getLogger(SimpleThread2.class);

	@Override
	public void run() {
		try {
			logger.info("线程 SimpleThread2 被调用！");
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

控制台打印：

14:06:05.009 [simple-threadpool-1] INFO com.c3stones.simple.SimpleThread2 - 线程 SimpleThread2 被调用！
14:06:05.009 [simple-threadpool-3] INFO com.c3stones.simple.SimpleThread2 - 线程 SimpleThread2 被调用！
14:06:05.010 [simple-threadpool-0] INFO com.c3stones.simple.SimpleThread1 - 线程 SimpleThread1 被调用！
14:06:05.009 [simple-threadpool-4] INFO com.c3stones.simple.SimpleThread1 - 线程 SimpleThread1 被调用！
14:06:05.009 [simple-threadpool-2] INFO com.c3stones.simple.SimpleThread1 - 线程 SimpleThread1 被调用！
14:06:06.017 [simple-threadpool-3] INFO com.c3stones.simple.SimpleThread2 - 线程 SimpleThread2 被调用！
14:06:06.018 [simple-threadpool-1] INFO com.c3stones.simple.SimpleThread1 - 线程 SimpleThread1 被调用！
14:06:07.017 [simple-threadpool-3] INFO com.c3stones.simple.SimpleThread2 - 线程 SimpleThread2 被调用！
14:06:08.017 [simple-threadpool-3] INFO com.c3stones.simple.SimpleThread1 - 线程 SimpleThread1 被调用！
14:06:08.018 [simple-threadpool-2] INFO com.c3stones.simple.SimpleThread2 - 线程 SimpleThread2 被调用！

  问题：
  (1) 当进程被异常关闭，会导致存储在线程池或者队列的线程丢失。
  (2) 但是消息队列中的消息不会因为JVM进程关闭而丢失，依然存储在消息队列所在服务器上。

3. 快速入门

• 开启异步支持

/**
 * 启动类
 * 
 * @author CL
 *
 */
@SpringBootApplication
@EnableAsync // 开启异步支持
public class Application {

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

}

• 编写业务代码

import org.springframework.stereotype.Service;

import lombok.extern.slf4j.Slf4j;

/**
 * 快速入门
 * 
 * @author CL
 *
 */
@Service
@Slf4j
public class TestQuickService {

	@SneakyThrows
	public String get() {
		log.info("调用TestQuickService.get()！");
		Thread.sleep(2000);
		return "get";
	}

	@SneakyThrows
	public String get2() {
		log.info("调用TestQuickService.get2()！");
		Thread.sleep(5000);
		return "get2";
	}
	
}

• 测试同步调用

/**
 * 测试快速入门
 * 
 * @author CL
 *
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
@Slf4j
public class TestQuick {

	@Autowired
	private TestQuickService quickService;

	/**
	 * 测试同步调用
	 */
	@Test
	public void testSync() {
		LocalDateTime startTime = LocalDateTime.now();

		quickService.get();
		quickService.get2();

		LocalDateTime endTime = LocalDateTime.now();

		log.info("同步调用，总耗时：" + Duration.between(startTime, endTime).toMillis() + " ms");
	}

}

控制台打印：

2020-05-28 16:39:36.853  INFO 6220 --- [           main] com.c3stones.quick.TestQuickService      : 调用TestQuickService.get()！
2020-05-28 16:39:38.853  INFO 6220 --- [           main] com.c3stones.quick.TestQuickService      : 调用TestQuickService.get2()！
2020-05-28 16:39:43.857  INFO 6220 --- [           main] com.c3stones.test.TestQuick              : 同步调用，总耗时：7000 ms

  可以看出总耗时为两个业务方法的总耗时，并且在主线程中执行。

• 异步调用，添加 @Async 注解

@Async
public String asyncGet() {
	return this.get();
}

@Async
public String asyncGet2() {
	return this.get2();
}

• 测试异步调用

/**
 * 测试异步调用
 */
@Test
public void testAsync() {
	LocalDateTime startTime = LocalDateTime.now();

	quickService.asyncGet();
	quickService.asyncGet2();

	LocalDateTime endTime = LocalDateTime.now();

	log.info("异步调用，总耗时：" + Duration.between(startTime, endTime).toMillis() + " ms");
}

控制台打印：

2020-05-28 17:11:10.550  INFO 908 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-05-28 17:11:10.563  INFO 908 --- [           main] com.c3stones.test.TestQuick              : 异步调用，总耗时：45 ms
2020-05-28 17:11:10.574  INFO 908 --- [         task-2] com.c3stones.quick.TestQuickService      : 调用TestQuickService.get2()！
2020-05-28 17:11:10.575  INFO 908 --- [         task-1] com.c3stones.quick.TestQuickService      : 调用TestQuickService.get()！
2020-05-28 17:11:10.587  INFO 908 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'

  可以看出总耗时不受业务方法时间的影响，并且都在异步线程池中执行的。
  注意：实际两个方法并没有执行完成。

4. 异步调用但主线程阻塞

  实际场景中肯定不可能出现上述的问题。肯定希望既能异步调用，并且主线程能阻塞，直到方法执行完成。

• 改进代码，返回Futrue对象

@Async
public Future<String> asyncFutureGet() {
	return AsyncResult.forValue(this.get());
}

@Async
public Future<String> asyncFutureGet2() {
	return AsyncResult.forValue(this.get2());
}

• 测试异步调用并阻塞主线程

/**
 * 测试异步调用并阻塞主线程
 */
@Test
@SneakyThrows
public void testAsyncFuture() {
	LocalDateTime startTime = LocalDateTime.now();

	// 执行
	Future<String> getFuture = quickService.asyncFutureGet();
	Future<String> get2Future = quickService.asyncFutureGet2();

	// 阻塞等待执行结果
	getFuture.get();
	get2Future.get();

	LocalDateTime endTime = LocalDateTime.now();

	log.info("异步调用，总耗时：" + Duration.between(startTime, endTime).toMillis() + " ms");
}

控制台打印：

2020-05-28 17:14:57.434  INFO 5784 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-05-28 17:14:57.468  INFO 5784 --- [         task-2] com.c3stones.quick.TestQuickService      : 调用TestQuickService.get2()！
2020-05-28 17:14:57.470  INFO 5784 --- [         task-1] com.c3stones.quick.TestQuickService      : 调用TestQuickService.get()！
2020-05-28 17:15:02.474  INFO 5784 --- [           main] com.c3stones.test.TestQuick              : 异步调用，总耗时：5066 ms
2020-05-28 17:15:02.511  INFO 5784 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'

  可以看出总耗时由耗时较长的方法决定，并且在异步线程池中执行。

5. Spring Task配置

• 编写配置文件application.yml

spring:
   task: # Spring执行器配置，对应TaskExecutionProperties配置类。对于Spring异步任务，会使用该执行器。
      execution:
         thread-name-prefix: task- # 线程池的线程名的前缀。默认为 task- ，根据自己应用来设置。
         pool: # 线程池相关
            core-size: 8 # 核心线程数，线程池创建时初始化的线程数。默认为 8。
            max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE。
            keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒。
            queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE。
            allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true。
         shutdown:
            await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true。
            await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0。

  在spring.task.execution配置项Spring Task调度任务的配置对应TaskExecutionProperties配置类。
  Spring Boot TaskExecutionAutoConfiguration自动化配置类，实现Spring Task的自动配置，创建ThreadPoolTaskExecutor 基于线程池的任务执行器。本质上ThreadPoolTaskExecutor是基于 ThreadPoolExecutor的封装，主要增加了提交任务，返回ListenableFuture 对象的功能。
  spring.task.execution.shutdown配置项是为了实现Spring Task异步任务的优雅关闭。
  在异步任务执行过程中，如果应用开始关闭，把异步任务需要使用到的Spring Bean一并销毁，例如数据库连接池等，但是异步任务还在执行中，当需要访问数据库时，就会导致报错。所以通过配置await-termination = true来实现应用关闭时，等待异步任务执行完成。这样应用在关闭时Spring 会优先等待ThreadPoolTaskSchedule 执行完任务之后，再开始Spring Bean的销毁。
  同时，又考虑到我们不可能无限等待异步任务全部执行结束，因此可以配置await-termination-period = 60，等待任务完成的最大时长，单位为秒。具体设置需与根据业务场景决定。

6. 异步回调

• AsyncResult类，异步结果
• ListenableFuture接口，监听Future
• Future接口，返回异步计算结果
    org.springframework.scheduling.annotation.AsyncResult<V>类实现了org.springframework.util.concurrent.ListenableFuture<T>接口，org.springframework.util.concurrent.ListenableFuture<T>又继承了java.util.concurrent.Future<V>接口。
    Future表示一个异步计算任务，当任务完成时得到计算结果。如果希望执行完成后马上得到结果，则需要另一个线程不断的查询监控状态，毋庸置疑增加了一定的复杂度。ListenableFuture对java原生Future进行扩展增强，其实就是监听Future是否执行完成，并自动调用回调方法，减少并发程序的复杂度。
  源码分析：

public interface Future<V> {

	/**
	 * 1. 如果任务没有开始，mayInterruptIfRunning = true/false，则直接返回false；<br/>
	 * 2. 如果任务正在执行中，mayInterruptIfRunning = true，则试图中断任务，如果成功中断，则返回true；<br/>
	 * 3. 如果任务正在执行中，mayInterruptIfRunning = false，则不会对执行中的线程产生影响，则直接返回false;<br/>
	 * 4. 如果任务已经完成，mayInterruptIfRunning = true/false，则直接返回false；<br/>
	 * 
	 * @param mayInterruptIfRunning 是否中断执行中的线程
	 * @return true/false
	 */
	boolean cancel(boolean mayInterruptIfRunning);

	/**
	 * 如果任务被中断，则返回true
	 * 
	 * @return true/false
	 */
	boolean isCancelled();

	/**
	 * 如果任务已经完成，无论是正常结束或是中端断都返回 true
	 * 
	 * @return true/false
	 */
	boolean isDone();

	/**
	 * 获得异步结果，如果任务还在执行中，则阻塞直到异步任务完成
	 * 
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	V get() throws InterruptedException, ExecutionException;

	/**
	 * 获得异步结果，如果任务还在执行中，则阻塞直到异步任务完成<br/>
	 * 但是有时间限制，如果阻塞时间超过设定的timeout，则抛出异常。
	 * 
	 * @param  timeout 超时时间
	 * @param  unit    时间单位枚举
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

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

	/**
	 * 回调方法，处理异常结果（不论成功还是异常）
	 * 
	 * @param callback
	 */
	void addCallback(ListenableFutureCallback<? super T> callback);

	/**
	 * 回调方法，分别处理成功和异常结果
	 * 
	 * @param successCallback 成功回调
	 * @param failureCallback 异常回调
	 */
	void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback);

	/**
	 * 转换成JDK1.8提供的CompletableFuture类，该类提供了非常强大的Future扩展功能
	 * 
	 * @return
	 */
	default CompletableFuture<T> completable() {
		CompletableFuture<T> completable = new DelegatingCompletableFuture<>(this);
		addCallback(completable::complete, completable::completeExceptionally);
		return completable;
	}

}

public class AsyncResult<V> implements ListenableFuture<V> {

	@Nullable
	private final V value;

	@Nullable
	private final Throwable executionException;

	public AsyncResult(@Nullable V value) {
		this(value, null);
	}

	private AsyncResult(@Nullable V value, @Nullable Throwable ex) {
		this.value = value;
		this.executionException = ex;
	}

	/**
	 * AsyncResult表示异常结果，则表示取消失败
	 * 
	 * @param mayInterruptIfRunning 是否中断执行中的线程
	 * @return false
	 */
	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		return false;
	}

	/**
	 * AsyncResult表示异常结果，则表示取消失败
	 * 
	 * @return false
	 */
	@Override
	public boolean isCancelled() {
		return false;
	}

	/**
	 * AsyncResult表示异常结果，则表示执行完成
	 * 
	 * @return true
	 */
	@Override
	public boolean isDone() {
		return true;
	}

	/**
	 * 获得异步结果，如果任务还在执行中，则阻塞直到异步任务完成
	 * 
	 * @return
	 * @throws ExecutionException
	 */
	@Override
	@Nullable
	public V get() throws ExecutionException {
		if (this.executionException != null) {
			throw (this.executionException instanceof ExecutionException ? (ExecutionException) this.executionException
					: new ExecutionException(this.executionException));
		}
		return this.value;
	}

	/**
	 * 获得异步结果，如果任务还在执行中，则阻塞直到异步任务完成<br/>
	 * 但是有时间限制，如果阻塞时间超过设定的timeout，则抛出异常。
	 * 
	 * @param timeout 超时时间
	 * @param unit    时间单位枚举
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	@Override
	@Nullable
	public V get(long timeout, TimeUnit unit) throws ExecutionException {
		return get();
	}

	/**
	 * 回调方法，处理成功和异常的结果
	 * 
	 * @param callback
	 */
	@Override
	public void addCallback(ListenableFutureCallback<? super V> callback) {
		addCallback(callback, callback);
	}

	/**
	 * 回调方法，分别处理成功和异常的结果<br/>
	 * catch中忽略了回调中发生的异常。当多个对调方法执行时，不会因为某一个回调方法异常而影响其他的回调方法
	 * 
	 * @param successCallback 成功回调
	 * @param failureCallback 异常回调
	 */
	@Override
	public void addCallback(SuccessCallback<? super V> successCallback, FailureCallback failureCallback) {
		try {
			if (this.executionException != null) {
				failureCallback.onFailure(exposedException(this.executionException));
			} else {
				successCallback.onSuccess(this.value);
			}
		} catch (Throwable ex) {
			// Ignore
		}
	}

	/**
	 * 返回CompletableFuture
	 * 
	 * @return
	 */
	@Override
	public CompletableFuture<V> completable() {
		if (this.executionException != null) {
			CompletableFuture<V> completable = new CompletableFuture<>();
			completable.completeExceptionally(exposedException(this.executionException));
			return completable;
		} else {
			return CompletableFuture.completedFuture(this.value);
		}
	}

	/**
	 * 执行成功，返回ListenableFuture
	 * 
	 * @param <V>
	 * @param value
	 * @return
	 */
	public static <V> ListenableFuture<V> forValue(V value) {
		return new AsyncResult<>(value, null);
	}

	/**
	 * 执行失败，返回ListenableFuture，执行回调方法
	 * 
	 * @param <V>
	 * @param ex
	 * @return
	 */
	public static <V> ListenableFuture<V> forExecutionException(Throwable ex) {
		return new AsyncResult<>(null, ex);
	}

	/**
	 * 获得具体的异常
	 * 
	 * @param original
	 * @return
	 */
	private static Throwable exposedException(Throwable original) {
		if (original instanceof ExecutionException) {
			Throwable cause = original.getCause();
			if (cause != null) {
				return cause;
			}
		}
		return original;
	}

}

• ListenableFutureTask类，异步增强ListenableFuture
    org.springframework.util.concurrent.ListenableFutureTask<T>类继承了java.util.concurrent.FutureTask<T>类,实现了org.springframework.util.concurrent.ListenableFuture<T>接口。

public class ListenableFutureTask<T> extends FutureTask<T> implements ListenableFuture<T> {

	/**
	 * 暂存回调
	 */
	private final ListenableFutureCallbackRegistry<T> callbacks = new ListenableFutureCallbackRegistry<>();

	public ListenableFutureTask(Callable<T> callable) {
		super(callable);
	}

	public ListenableFutureTask(Runnable runnable, @Nullable T result) {
		super(runnable, result);
	}

	/**
	 * 回调方法，处理成功和异常的结果
	 * 
	 * @param callback
	 */
	@Override
	public void addCallback(ListenableFutureCallback<? super T> callback) {
		this.callbacks.addCallback(callback);
	}

	/**
	 * 回调方法，分别处理成功和异常的结果<br/>
	 * catch中忽略了回调中发生的异常。当多个对调方法执行时，不会因为某一个回调方法异常而影响其他的回调方法
	 * 
	 * @param successCallback 成功回调
	 * @param failureCallback 异常回调
	 */
	@Override
	public void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback) {
		this.callbacks.addSuccessCallback(successCallback);
		this.callbacks.addFailureCallback(failureCallback);
	}

	/**
	 * 返回CompletableFuture
	 * 
	 * @return
	 */
	@Override
	public CompletableFuture<T> completable() {
		CompletableFuture<T> completable = new DelegatingCompletableFuture<>(this);
		this.callbacks.addSuccessCallback(completable::complete);
		this.callbacks.addFailureCallback(completable::completeExceptionally);
		return completable;
	}

	/**
	 * 重写FutureTask中的done方法
	 */
	@Override
	protected void done() {
		Throwable cause;
		try {
			// 获得执行结果
			T result = get();
			// 执行成功，执行成功回调方法
			this.callbacks.success(result);
			return;
		} catch (InterruptedException ex) { // 中断异常，并返回
			Thread.currentThread().interrupt();
			return;
		} catch (ExecutionException ex) { // 获得具体的异常，并设置到cause中
			cause = ex.getCause();
			if (cause == null) {
				cause = ex;
			}
		} catch (Throwable ex) { // 设置到cause中
			cause = ex;
		}
		// 执行异常，执行异常回调方法
		this.callbacks.failure(cause);
	}

}

7. 测试异步回调

• 编写业务代码

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

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

/**
 * 测试回调
 * 
 * @author CL
 *
 */
@Service
@Slf4j
public class TestCallbackService {

	@SneakyThrows
	public String get() {
		log.info("调用TestCallbackService.get()！");
		Thread.sleep(2000);
		return "get";
	}

	@Async
	public ListenableFuture<String> asyncCallback() {
		try {
			return AsyncResult.forValue(this.get());
		} catch (Throwable ex) {
			return AsyncResult.forExecutionException(ex);
		}
	}
}

• 测试异步回调

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

/**
 * 测试异步回调
 * 
 * @author CL
 *
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
@Slf4j
public class TestCallback {

	@Autowired
	private TestCallbackService callbackService;

	/**
	 * 测试异步回调
	 */
	@Test
	@SneakyThrows
	public void testCallback() {
		LocalDateTime startTime = LocalDateTime.now();

		ListenableFuture<String> listenableFuture = callbackService.asyncCallback();
		log.info("返回类型为：" + listenableFuture.getClass().getSimpleName());

		// 分别增加成功和异常的回调
		listenableFuture.addCallback(new SuccessCallback<String>() {

			@Override
			public void onSuccess(String result) {
				log.info("执行成功！");
			}

		}, new FailureCallback() {

			@Override
			public void onFailure(Throwable ex) {
				log.error("执行异常：" + ex);
			}

		});
		// 增加统一的成功和异常回调
		listenableFuture.addCallback(new ListenableFutureCallback<String>() {

			@Override
			public void onSuccess(String result) {
				log.info("执行成功！");
			}

			@Override
			public void onFailure(Throwable ex) {
				log.error("执行异常：" + ex);
			}

		});
		// 阻塞主线程
		listenableFuture.get();

		LocalDateTime endTime = LocalDateTime.now();

		log.info("异步调用，总耗时：" + Duration.between(startTime, endTime).toMillis() + " ms");
	}

}

控制台打印：

2020-05-29 11:56:24.815  INFO 17920 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-05-29 11:56:24.822  INFO 17920 --- [           main] com.c3stones.test.TestCallback           : 返回类型为：ListenableFutureTask
2020-05-29 11:56:24.835  INFO 17920 --- [         task-1] c.c3stones.callback.TestCallbackService  : 调用TestCallbackService.get()！
2020-05-29 11:56:26.838  INFO 17920 --- [         task-1] com.c3stones.test.TestCallback           : 执行成功！
2020-05-29 11:56:26.838  INFO 17920 --- [         task-1] com.c3stones.test.TestCallback           : 执行成功！
2020-05-29 11:56:26.843  INFO 17920 --- [           main] com.c3stones.test.TestCallback           : 异步调用，总耗时：2055 ms
2020-05-29 11:56:26.874  INFO 17920 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'

8. 全局异步异常处理器

  实现AsyncUncaughtExceptionHandler接口即可实现异步调用的统一异常处理。
  注意：AsyncUncaughtExceptionHandler返回非Future类型 的异步调用方法。因此返回Future类型的异步方法只能使用回调方法处理。
  从AsyncExecutionAspectSupport的handleError方法可以看出上述结论。

protected void handleError(Throwable ex, Method method, Object... params) throws Exception {
        // 返回类型为Future，直接抛出异常
		if (Future.class.isAssignableFrom(method.getReturnType())) {
			ReflectionUtils.rethrowException(ex);
		}
		// 否则，交给AsyncUncaughtExceptionHandler来处理
		else {
			// Could not transmit the exception to the caller with default executor
			try {
				this.exceptionHandler.obtain().handleUncaughtException(ex, method, params);
			}
			catch (Throwable ex2) {
				logger.warn("Exception handler for async method '" + method.toGenericString() +
						"' threw unexpected exception itself", ex2);
			}
		}
	}

• 编写全局异步异常处理器

import java.lang.reflect.Method;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

/**
 * 全局异步异常处理器
 * 
 * @author CL
 *
 */
@Component
@Slf4j
public class GlobalAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler {

	/**
	 * 处理未捕获的异步调用异常
	 */
	@Override
	public void handleUncaughtException(Throwable ex, Method method, Object... params) {
		log.error("方法：{}, 参数：{}，调用异常：{}", method, params, ex.getMessage());
	}

}

• 编写异步配置类

import java.util.concurrent.Executor;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;

import com.c3stones.handle.GlobalAsyncUncaughtExceptionHandler;

@Configuration
@EnableAsync // 开启异步支持（和Application启动类中保留一个即可，建议保留在此处）
public class AsyncConfig implements AsyncConfigurer {

	@Autowired
	private GlobalAsyncUncaughtExceptionHandler asyncUncaughtExceptionHandler;

	/**
	 * 返回 Spring Task异步任务的默认执行器。<br/>
	 * 返回了null，则使用TaskExecutionAutoConfiguration自动化配置类创建的ThreadPoolTaskExecutor任务执行器作为默认执行器
	 */
	@Override
	public Executor getAsyncExecutor() {
		return null;
	}

	/**
	 * 返回自定义的全局异步异常处理器
	 */
	@Override
	public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
		return asyncUncaughtExceptionHandler;
	}

}

• 测试全局异步异常处理

import java.time.Duration;
import java.time.LocalDateTime;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.c3stones.Application;
import com.c3stones.exceptionHandle.TestAsyncExceptionService;

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

/**
 * 测试异步异常处理
 * 
 * @author CL
 *
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
@Slf4j
public class TestAsyncException {

	@Autowired
	private TestAsyncExceptionService asyncExceptionService;

	/**
	 * 测试异步异常
	 */
	@Test
	@SneakyThrows
	public void testAsyncException() {
		LocalDateTime startTime = LocalDateTime.now();

		asyncExceptionService.asyncException();

		// 异步调用成功
		Thread.sleep(5000);

		LocalDateTime endTime = LocalDateTime.now();

		log.info("异步异常调用，总耗时：" + Duration.between(startTime, endTime).toMillis() + " ms");
	}

}

控制台打印：

2020-05-29 11:59:12.804  INFO 12724 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-05-29 11:59:12.821 ERROR 12724 --- [         task-1] .c.h.GlobalAsyncUncaughtExceptionHandler : 方法：public java.lang.String com.c3stones.exceptionHandle.TestAsyncExceptionService.asyncException(), 参数：[]，调用异常：TestAsyncExceptionService.asyncException抛出异常！
2020-05-29 11:59:17.817  INFO 12724 --- [           main] com.c3stones.test.TestAsyncException     : 异步异常调用，总耗时：5048 ms
2020-05-29 11:59:17.856  INFO 12724 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'

9. 自定义任务执行器

  使用Spring Boot TaskExecutionAutoConfiguration 自动化配置类，实现自动配置ThreadPoolTaskExecutor任务执行器。

• 创建Maven工程
  
• 修改pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.c3stones</groupId>
	<artifactId>spring-boot-async-demo2</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>spring-boot-async-demo2</name>
	<description>Spring Boot Async Demo2</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.2.6.RELEASE</version>
		<relativePath />
	</parent>

	<properties>
		<java.version>1.8</java.version>
		<maven-jar-plugin.version>3.1.1</maven-jar-plugin.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>
	
</project>

• 编写application.yml

spring:
   task: # Spring执行器配置，对应TaskExecutionProperties配置类。对于Spring异步任务，会使用该执行器。
      execution-one:
         thread-name-prefix: task-one- # 线程池的线程名的前缀。默认为 task- ，根据自己应用来设置。
         pool: # 线程池相关
            core-size: 8 # 核心线程数，线程池创建时初始化的线程数。默认为 8。
            max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE。
            keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒。
            queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE。
            allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true。
         shutdown:
            await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true。
            await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0。
      execution-two:
         thread-name-prefix: task-two- # 线程池的线程名的前缀。默认为 task- ，根据自己应用来设置。
         pool: # 线程池相关
            core-size: 8 # 核心线程数，线程池创建时初始化的线程数。默认为 8。
            max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE。
            keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒。
            queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE。
            allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true。
         shutdown:
            await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true。
            await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0。

• 配置执行器

import org.springframework.boot.autoconfigure.task.TaskExecutionProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.task.TaskExecutorBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * 异步配置
 * 
 * @author CL
 *
 */
@Configuration
@EnableAsync // 开启异步的支持
public class AsyncConfig {

	/**
	 * 执行器1
	 */
	public static final String EXECUTOR_ONE = "executor-one";
	/**
	 * 执行器2
	 */
	public static final String EXECUTOR_TWO = "executor-two";

	/**
	 * 配置执行器1
	 * 
	 * @author CL
	 *
	 */
	@Configuration
	public static class ExecutorOneConfiguration {

		@Bean(name = EXECUTOR_ONE + "-properties")
		@ConfigurationProperties(prefix = "spring.task.execution-one")
		@Primary
		public TaskExecutionProperties taskExecutionProperties() {
			return new TaskExecutionProperties();
		}

		@Bean(name = EXECUTOR_ONE)
		public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
			// 通过TaskExecutorBuilder对象创建ThreadPoolTaskExecutor
			TaskExecutorBuilder builder = createTaskExecutorBuilder(this.taskExecutionProperties());
			return builder.build();
		}

	}

	/**
	 * 配置执行器2
	 * 
	 * @author CL
	 *
	 */
	@Configuration
	public static class ExecutorTwoConfiguration {

		@Bean(name = EXECUTOR_TWO + "-properties")
		@ConfigurationProperties(prefix = "spring.task.execution-two")
		public TaskExecutionProperties taskExecutionProperties() {
			return new TaskExecutionProperties();
		}

		@Bean(name = EXECUTOR_TWO)
		public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
			// 通过TaskExecutorBuilder对象创建ThreadPoolTaskExecutor
			TaskExecutorBuilder builder = createTaskExecutorBuilder(this.taskExecutionProperties());
			return builder.build();
		}

	}

	/**
	 * 创建TaskExecutorBuilder
	 * 
	 * @param properties 配置，从配置文件读取
	 * @return
	 */
	private static TaskExecutorBuilder createTaskExecutorBuilder(TaskExecutionProperties properties) {
		TaskExecutionProperties.Pool pool = properties.getPool();
		TaskExecutorBuilder builder = new TaskExecutorBuilder();
		// 配置属性，与配置文件对应
		// 其它基本属性
		builder = builder.threadNamePrefix(properties.getThreadNamePrefix());
		// Pool 属性
		builder = builder.corePoolSize(pool.getCoreSize());
		builder = builder.maxPoolSize(pool.getMaxSize());
		builder = builder.keepAlive(pool.getKeepAlive());
		builder = builder.queueCapacity(pool.getQueueCapacity());
		builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());
		// Shutdown 属性
		TaskExecutionProperties.Shutdown shutdown = properties.getShutdown();
		builder = builder.awaitTermination(shutdown.isAwaitTermination());
		builder = builder.awaitTerminationPeriod(shutdown.getAwaitTerminationPeriod());
		return builder;
	}

}

• 编写业务逻辑，指定执行器

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.c3stones.config.AsyncConfig;

import lombok.extern.slf4j.Slf4j;

/**
 * 自定义执行器
 * 
 * @author CL
 *
 */
@Service
@Slf4j
public class TestExecutorService {

	/**
	 * 指定执行器1
	 * 
	 * @return
	 */
	@Async(AsyncConfig.EXECUTOR_ONE)
	public String get() {
		log.info("调用TestExecutorService.get()！");
		return "get";
	}

	/**
	 * 指定执行器2
	 * 
	 * @return
	 */
	@Async(AsyncConfig.EXECUTOR_TWO)
	public String get2() {
		log.info("调用TestExecutorService.get2()！");
		return "get2";
	}

}

• 编写启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * 启动类
 * 
 * @author CL
 *
 */
@SpringBootApplication
public class Application {

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

}

• 测试自定义执行器

import java.time.Duration;
import java.time.LocalDateTime;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.c3stones.Application;
import com.c3stones.service.TestExecutorService;

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

/**
 * 测试自定义执行器
 * 
 * @author CL
 *
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
@Slf4j
public class TestExecutor {

	@Autowired
	private TestExecutorService executorService;

	/**
	 * 自定义执行器
	 */
	@Test
	@SneakyThrows
	public void testExecute() {
		LocalDateTime startTime = LocalDateTime.now();

		executorService.get();
		executorService.get2();

		Thread.sleep(2000);

		LocalDateTime endTime = LocalDateTime.now();

		log.info("总耗时：" + Duration.between(startTime, endTime).toMillis() + " ms");
	}

}

控制台打印：

2020-05-29 13:13:32.070  INFO 15564 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'executor-one'
2020-05-29 13:13:32.114  INFO 15564 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'executor-two'
2020-05-29 13:13:32.201  INFO 15564 --- [           main] com.c3stones.test.TestExecutor           : Started TestExecutor in 1.427 seconds (JVM running for 2.584)
2020-05-29 13:13:32.576  INFO 15564 --- [     task-two-1] c.c3stones.service.TestExecutorService   : 调用TestExecutorService.get2()！
2020-05-29 13:13:32.577  INFO 15564 --- [     task-one-1] c.c3stones.service.TestExecutorService   : 调用TestExecutorService.get()！
2020-05-29 13:13:34.554  INFO 15564 --- [           main] com.c3stones.test.TestExecutor           : 总耗时：2006 ms
2020-05-29 13:13:34.581  INFO 15564 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'executor-two'
2020-05-29 13:13:34.581  INFO 15564 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'executor-one'

10. 项目地址

  spring-boot-async