在Spring中，实现异步调用主要有三种方式：
方式一：注解方式
要开启异步支持，首先得在Spring Boot入口类上加上@EnableAsync注解：
@SpringBootApplication
@EnableAsync
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
新建service包，并创建TestService：
@Service
public class TestService {
private Logger logger = LoggerFactory.getLogger(this.getClass());
@Async
public void asyncMethod() {
sleep();
logger.info("异步方法内部线程名称：{}", Thread.currentThread().getName());
}
public void syncMethod() {
sleep();
}
private void sleep() {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
上面的Service中包含一个异步方法asyncMethod（开启异步支持后，只需要在方法上加上@Async注解便是异步方法了）和同步方法syncMethod。sleep方法用于让当前线程阻塞2秒钟。
因为异步的原因，程序并没有被sleep方法阻塞，这就是异步调用的好处。同时异步方法内部会新启一个线程来执行
默认情况下的异步线程池配置使得线程不能被重用，每次调用异步方法都会新建一个线程，我们可以自己定义异步线程池来优化。
@Configuration
public class AsyncPoolConfig {
@Bean
public ThreadPoolTaskExecutor asyncThreadPoolTaskExecutor(){
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(20);
executor.setMaxPoolSize(200);
executor.setQueueCapacity(25);
executor.setKeepAliveSeconds(200);
executor.setThreadNamePrefix("asyncThread");
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(60);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
要使用该线程池，只需要在@Async注解上指定线程池Bean名称即可：
@Service
public class TestService {
......
@Async("asyncThreadPoolTaskExecutor")
public void asyncMethod() {
......
}
......
}
处理异步回调
如果异步方法具有返回值的话，需要使用Future来接收回调值。我们修改TestService的asyncMethod方法，给其添加返回值：
@Async("asyncThreadPoolTaskExecutor")
public Future<String> asyncMethod() {
sleep();
logger.info("异步方法内部线程名称：{}", Thread.currentThread().getName());
return new AsyncResult<>("hello async");
}
Future接口的get方法用于获取异步调用的返回值。
通过返回结果我们可以看出Future的get方法为阻塞方法，只有当异步方法返回内容了，程序才会继续往下执行。get还有一个get(long timeout, TimeUnit unit)重载方法，我们可以通过这个重载方法设置超时时间，即异步方法在设定时间内没有返回值的话，直接抛出java.util.concurrent.TimeoutException异常。
比如设置超时时间为60秒：
String result = stringFuture.get(60, TimeUnit.SECONDS);
方式二：内置线程池方式
可以使用Spring内置的线程池来实现异步调用，比如ThreadPoolTaskExecutor 和SimpleAsyncTaskExecutor。Spring提供了许多TaskExecutor的内置实现。下面简单介绍5种内置的线程池。
1）SimpleAsyncTaskExecutor：它不会复用线程，每次调用都是启动一个新线程。
2）ConcurrentTaskExecutor：它是Java API中Executor实例的适配器。
3）ThreadPoolTaskExecutor：这个线程池是最常用的。它公开了用于配置的bean属性，并将它包装在TaskExecutor中。
4）WorkManagerTaskExecutor：它基于CommonJ WorkManager来实现的，并且是在Spring上下文中的WebLogic或WebSphere中设置CommonJ线程池的工具类。
5）DefaultManagedTaskExecutor：主要用于支持JSR-236兼容的运行时环境，它是使用JNDI获得ManagedExecutorService，作为CommonJ WorkManager的替代方案。
通常情况下，ThreadPoolTaskExecuto最为常用，只要当ThreadPoolTaskExecutor不能满足需求时，可以使用ConcurrentTaskExecutor。如果在代码中声明了多个线程池，Spring会默认按照以下搜索顺序来调用线程池：
第一步，检查上下文中的唯一TaskExecutor Bean。
第二步，检查名为“ taskExecutor”的Executor Bean。
第三步，以上都无法无法处理，就会使用SimpleAsyncTaskExecutor来执行。
示例1：
CompletableFuture.runAsync()来完成异步任务（标准版）
（1）配置线程池
/**
* int corePoolSize,
* int maximumPoolSize,
* long keepAliveTime,
* TimeUnit unit,
* BlockingQueue<Runnable> workQueue,
* ThreadFactory threadFactory,
* RejectedExecutionHandler handler
* @return
*/
@Bean
public ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties pool) {
return new ThreadPoolExecutor(pool.getCoreSize(),
pool.getMaxSize(),
pool.getKeepAliveTime(),
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(100000),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
}
（2）线程池参数配置类
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
my.thread.core-size=20
my.thread.max-size=200
my.thread.keep-alive-time=10
*/
@ConfigurationProperties(prefix = "my.thread")
@Component
@Data
public class ThreadPoolConfigProperties {
//核心线程数
private Integer coreSize;
//最大线程数
private Integer maxSize;
//空余线程的存活时间
private Integer keepAliveTime;
}
（3）测试异步任务
@Autowired
private ThreadPoolExecutor executor;
//在这里开启一个异步任务，提交给线程池，runAsync()方法没有返回值，需要有返回值的可使用supplyAsync()方法
@Test
void testCompletableFuture() {
CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {
int result = 0;
for (int i = 0; i <= 100; i++) {
result += i;
}
System.out.println(result);
}, executor);
}
（4）关于CompletableFuture的其他相关用法
4.1 CompletableFuture的**get()**方法可以获取异步的结果，get方法是一个阻塞式等待的方法，也即get方法会等待异步任务的完成
CompletableFuture<AtomicInteger> completableFuture2 = CompletableFuture.supplyAsync(() -> {
for (int i = 0; i <= 100; i++) {
sum2.addAndGet(i);
}
return sum2;
}, executor);
//获取异步结果
AtomicInteger integer = completableFuture2.get();
4.2 allOf : 等待所有任务完成完成
AtomicInteger sum = new AtomicInteger();
AtomicInteger sum2 = new AtomicInteger();
CompletableFuture<AtomicInteger> completableFuture1 = CompletableFuture.supplyAsync(() -> {
for (int i = 0; i <= 100; i++) {
sum.addAndGet(i);
}
return sum;
}, executor);
CompletableFuture<AtomicInteger> completableFuture2 = CompletableFuture.supplyAsync(() -> {
for (int i = 0; i <= 100; i++) {
sum2.addAndGet(i);
}
return sum2;
}, executor);
AtomicInteger integer = completableFuture2.get();
//allOf : 等待所有任务完成完成，注意get方法，是阻塞式等待，等待上面的异步任务都完成
CompletableFuture.allOf(completableFuture1,completableFuture2).get();
//获取异步结果
AtomicInteger atomicInteger1 = completableFuture1.get();
AtomicInteger atomicInteger2 = completableFuture2.get();
System.out.println("结果是--->"+atomicInteger1.addAndGet(atomicInteger2.intValue()));
异步任务完成时，whenComplete，exceptionally
CompletableFuture<AtomicInteger> completableFuture3 = CompletableFuture.supplyAsync(() -> {
for (int i = 0; i <= 10; i++) {
sum2.addAndGet(i);
}
return sum2;
}, executor).whenComplete((res, exception) -> {
//当出现异常，可以拿到异常信息，但是无法修改返回数据
System.out.println("结果是：" + res + ",异常：" + exception);
}).exceptionally(throwable -> {
//可以感知异常，同时返回默认值
return new AtomicInteger(10);
});
4.4 handle，方法完成后的后续处理
CompletableFuture<Integer> completableFuture4 = CompletableFuture.supplyAsync(() -> {
int i = 10 / 2;
return i;
}, executor).handle((res, throwable) -> {
//res 为结果，throwable 为异常
if (res != null) {
return res * 2;
}
if (throwable != null) {
return -1;
}
return 0;
});
System.out.println("completableFuture4--结果是："+completableFuture4.get());
4.5 异步任务串行化
/**
* 异步任务串行化
* thenAcceptAsync 可以接收上一步获取的结果，但是无返回值
* thenApplyAsync 可以接收上一步获取的结果，有返回值
*/
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
int i = 10 / 2;
return i;
}, executor).thenApplyAsync(res -> {
//res为上一步的结果
return res * 2;
}, executor).thenAcceptAsync((res) -> {
System.out.println("hello ...thenAcceptAsync");
}, executor);
示例2：
CompletableFuture.runAsync()来完成异步任务（简易版）
（1）初始化池
private ExecutorService executor = Executors.newFixedThreadPool(5);
（2）使用
@ApiOperation(value = "领实物奖")
@PostMapping(Routes.API_HELP_OFFLINE_PRIZE)
public Response getOfflinePrize(@Valid @RequestBody GetOfflinePrizeReq req, HttpServletRequest request) {
try {
UserInfo user = getUser(req.getToken(), request);
CompletableFuture.runAsync(() -> {
record.setPrizeName(prizes.get(0).getPrizeName());
helpFacadeService.sendPrizes(user, record, prizes);
helpFacadeService.recordPrizeAddress(req, user, prizes);
}, executor);
return ResponseUtils.buildResponse(null);
} catch (Exception e) {
log.error("[getPrize error] error: {}", e);
return ResponseUtils.buildErrorResponse(SYSTEM_ERROR, "领奖失败");
}
}
CompletableFuture提供了四个静态方法来创建一个异步操作：
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
这四个方法的区别是：
· runAsync() 以Runnable函数式接口类型为参数，没有返回结果，supplyAsync() 以Supplier函数式接口类型为参数，返回结果类型为U；Supplier接口的 get()是有返回值的(会阻塞)
· 使用没有指定Executor的方法时，内部使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。
· 默认情况下CompletableFuture会使用公共的ForkJoinPool线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置ForkJoinPool线程池的线程数）。如果所有CompletableFuture共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰
方式三：自定义线程池方式
可以通过实现AsyncConfigurer接口或者直接继承AsyncConfigurerSupport类来自定义线程池。但是非完全托管的Bean和完全托管的Bean实现方式有点小差异。
首先，来看非完全托管的Spring Bean，实现方式如代码所示：
@Configuration
@EnableAsync
public class AppConfig implements AsyncConfigurer {
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(7);
executor.setMaxPoolSize(42);
executor.setQueueCapacity(11);
executor.setThreadNamePrefix("MyExecutor-");
executor.initialize();//手动初始化
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new MyAsyncUncaughtExceptionHandler();
}
}
在这段代码中，ThreadPoolTaskExecutor不是完全托管的Spring bean。
然后，来看完全托管的Spring Bean，实现方式如代码所示
@Configuration
@EnableAsync
public class AppConfig implements AsyncConfigurer {
@Bean
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(7);
executor.setMaxPoolSize(42);
executor.setQueueCapacity(11);
executor.setThreadNamePrefix("MyExecutor-");
//executor.initialize();//不用手动调用
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new MyAsyncUncaughtExceptionHandler();
}
}
只要在异步方法上添加@Bean注解，不需要手动调用线程池的initialize()方法，在Bean在初始化之后会自动调用。需要注意的是，在同级类中直接调用异步方法无法实现异步。