Spring运用@Async注解,CompletableFuture基础用法

2019-11-18杂谈搜奇网45°c

A⁺ A^-

本文报告@Async注解，在Spring体系中的运用。本文仅申明@Async注解的运用划定规矩，关于道理，挪用逻辑，源码剖析，暂不引见。关于异步要领挪用，从Spring3最先供应了@Async注解，该注解能够被标注在要领上，以便异步地挪用该要领。挪用者将在挪用时马上返回，要领的现实执即将提交给Spring TaskExecutor的使命中，由指定的线程池中的线程实行。

在项目运用中，@Async挪用线程池，引荐运用自定义线程池的形式。自定义线程池经常使用计划：从新完成接口AsyncConfigurer。

简介

运用场景

同步：同步就是全部处置惩罚历程递次实行，当各个历程都实行终了，并返回效果。

异步：异步挪用则是只是发送了挪用的指令，挪用者无需守候被挪用的要领完整实行终了；而是继续实行下面的流程。比方，在某个挪用中，须要递次挪用 A, B, C三个历程要领；如他们都是同步挪用，则须要将他们都递次实行终了今后，方算作历程实行终了；如B为一个异步的挪用要领，则在实行完A今后，挪用B，并不守候B完成，而是实行最先挪用C，待C实行终了今后，就意味着这个历程实行终了了。在Java中，平常在处置惩罚相似的场景之时，都是基于建立自力的线程去完成响应的异步挪用逻辑，经由过程主线程和差别的营业子线程之间的实行流程，从而在启动自力的线程今后，主线程继续实行而不会发生阻滞守候的状况。

Spring 已完成的异常线程池

1. SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，默许每次挪用都邑建立一个新的线程。
2. SyncTaskExecutor：这个类没有完成异步挪用，只是一个同步操作。只适用于不须要多线程的处所。
3. ConcurrentTaskExecutor：Executor的适配类，不引荐运用。假如ThreadPoolTaskExecutor不满足要求时，才用斟酌运用这个类。
4. SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz运用，才须要运用此类。
5. ThreadPoolTaskExecutor ：最常运用，引荐。其实质是对java.util.concurrent.ThreadPoolExecutor的包装。

异步的要领有：

1. 最简朴的异步挪用，返回值为void
2. 带参数的异步挪用，异步要领能够传入参数
3. 存在返回值，常挪用返回Future

Spring中启用@Async

 1 // 基于Java设置的启用体式格局：
 2 @Configuration  
 3 @EnableAsync  
 4 public class SpringAsyncConfig { ... }  
 5 
 6 // Spring boot启用：
 7 @EnableAsync
 8 @EnableTransactionManagement
 9 public class SettlementApplication {
10     public static void main(String[] args) {
11         SpringApplication.run(SettlementApplication.class, args);
12     }
13 }

@Async运用默许线程池

Spring运用默许的线程池，指在@Async注解在运用时，不指定线程池的称号。检察源码，@Async的默许线程池为SimpleAsyncTaskExecutor。

无返回值挪用

基于@Async无返回值挪用，直接在运用类，运用要领（发起在运用要领）上，加上注解。若须要抛出异常，需手动new一个异常抛出。

 1  /**
 2      * 带参数的异步挪用 异步要领能够传入参数
 3      *  关于返回值是void，异常会被AsyncUncaughtExceptionHandler处置惩罚掉
 4      * @param s
 5      */
 6     @Async
 7     public void asyncInvokeWithException(String s) {
 8         log.info("asyncInvokeWithParameter, parementer={}", s);
 9         throw new IllegalArgumentException(s);
10     }

有返回值Future挪用

 1  /**
 2      * 异常挪用返回Future
 3      *  关于返回值是Future，不会被AsyncUncaughtExceptionHandler处置惩罚，须要我们在要领中捕捉异常并处置惩罚
 4      *  或许在挪用方在挪用Futrue.get时捕捉异常举行处置惩罚
 5      * 
 6      * @param i
 7      * @return
 8      */
 9     @Async
10     public Future<String> asyncInvokeReturnFuture(int i) {
11         log.info("asyncInvokeReturnFuture, parementer={}", i);
12         Future<String> future;
13         try {
14             Thread.sleep(1000 * 1);
15             future = new AsyncResult<String>("success:" + i);
16             throw new IllegalArgumentException("a");
17         } catch (InterruptedException e) {
18             future = new AsyncResult<String>("error");
19         } catch(IllegalArgumentException e){
20             future = new AsyncResult<String>("error-IllegalArgumentException");
21         }
22         return future;
23     }

有返回值CompletableFuture挪用

CompletableFuture 并不运用@Async注解，可到达挪用体系线程池处置惩罚营业的功用。

JDK5新增了Future接口，用于形貌一个异步盘算的效果。虽然 Future 以及相干运用要领供应了异步实行使命的才能，然则关于效果的猎取倒是很不轻易，只能经由过程壅塞或许轮询的体式格局获得使命的效果。壅塞的体式格局明显和我们的异步编程的初志相违犯，轮询的体式格局又会消耗无谓的 CPU 资本，而且也不能实时地获得盘算效果。

CompletionStage代表异步盘算历程当中的某一个阶段，一个阶段完成今后能够会触发别的一个阶段
一个阶段的盘算实行能够是一个Function，Consumer或许Runnable。比方：stage.thenApply(x -> square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println())
一个阶段的实行多是被单个阶段的完成触发，也多是由多个阶段一同触发

在Java8中，CompletableFuture供应了异常壮大的Future的扩大功用，能够协助我们简化异步编程的复杂性，而且供应了函数式编程的才能，能够经由过程回调的体式格局处置惩罚盘算效果，也供应了转换和组合 CompletableFuture 的要领。

它能够代表一个明白完成的Future，也有能够代表一个完成阶段（ CompletionStage ），它支撑在盘算完成今后触发一些函数或实行某些行动。
它完成了Future和CompletionStage接口

 1  /**
 2      * 数据查询线程池
 3      */
 4     private static final ThreadPoolExecutor SELECT_POOL_EXECUTOR = new ThreadPoolExecutor(10, 20, 5000,
 5             TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), new ThreadFactoryBuilder().setNameFormat("selectThreadPoolExecutor-%d").build());
 6 
 7 // tradeMapper.countTradeLog(tradeSearchBean)要领示意，猎取数目，返回值为int
 8  // 猎取总条数
 9         CompletableFuture<Integer> countFuture = CompletableFuture
10                 .supplyAsync(() -> tradeMapper.countTradeLog(tradeSearchBean), SELECT_POOL_EXECUTOR);
11 // 同步壅塞
12     CompletableFuture.allOf(countFuture).join();
13 // 猎取效果
14  int count = countFuture.get();

默许线程池的弊病

在线程池运用中，参考阿里巴巴java开辟范例：线程池不允许运用Executors去建立，不允许运用体系默许的线程池，引荐经由过程ThreadPoolExecutor的体式格局，如许的处置惩罚体式格局让开辟的工程师越发明白线程池的运转划定规矩，躲避资本耗尽的风险。Executors各个要领的弊病：

newFixedThreadPool和newSingleThreadExecutor：重要题目是聚集的要求处置惩罚行列能够会消耗异常大的内存，以至OOM。
newCachedThreadPool和newScheduledThreadPool：要题目是线程数最大数是Integer.MAX_VALUE，能够会建立数目异常多的线程，以至OOM。

@Async默许异步设置运用的是SimpleAsyncTaskExecutor，该线程池默许来一个使命建立一个线程，若体系中不停的建立线程，最终会致使体系占用内存太高，激发OutOfMemoryError毛病。针对线程建立题目，SimpleAsyncTaskExecutor供应了限流机制，经由过程concurrencyLimit属性来掌握开关，当concurrencyLimit>=0时开启限流机制，默许封闭限流机制即concurrencyLimit=-1，当封闭状况下，会不停建立新的线程来处置惩罚使命。基于默许设置，SimpleAsyncTaskExecutor并非严厉意义的线程池，达不到线程复用的功用。

@Async运用自定义线程池

自定义线程池，可对体系中线程池越发细粒度的掌握，轻易调解线程池大小设置，线程实行异常掌握和处置惩罚。自定义线程池有以下形式：

从新完成接口AsyncConfigurer
继续AsyncConfigurerSupport
设置由自定义的TaskExecutor替换内置的使命实行器

经由过程检察Spring源码关于@Async的默许挪用划定规矩，会优先查询源码中完成AsyncConfigurer这个接口的类，完成这个接口的类为AsyncConfigurerSupport。但默许设置的线程池和异步处置惩罚要领均为空，所以，无论是继续或许从新完成接口，都需指定一个线程池。

完成接口AsyncConfigurer

 1 @Configuration
 2 public class AsyncConfiguration implements AsyncConfigurer {
 3     @Bean("kingAsyncExecutor")
 4     public ThreadPoolTaskExecutor executor() {
 5         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 6         int corePoolSize = 10;
 7         executor.setCorePoolSize(corePoolSize);
 8         int maxPoolSize = 50;
 9         executor.setMaxPoolSize(maxPoolSize);
10         int queueCapacity = 10;
11         executor.setQueueCapacity(queueCapacity);
12         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
13         String threadNamePrefix = "kingDeeAsyncExecutor-";
14         executor.setThreadNamePrefix(threadNamePrefix);
15         executor.setWaitForTasksToCompleteOnShutdown(true);
16         // 运用自定义的跨线程的要求级别线程工场类
17         RequestContextThreadFactory threadFactory = RequestContextThreadFactory.getDefault();
18         executor.setThreadFactory(threadFactory);
19         int awaitTerminationSeconds = 5;
20         executor.setAwaitTerminationSeconds(awaitTerminationSeconds);
21         executor.initialize();
22         return executor;
23     }
24 
25     @Override
26     public Executor getAsyncExecutor() {
27         return executor();
28     }
29 
30     @Override
31     public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
32         return (ex, method, params) -> ErrorLogger.getInstance().log(String.format("实行异步使命'%s'", method), ex);
33     }
34 }

继续AsyncConfigurerSupport

@Configuration  
@EnableAsync  
class SpringAsyncConfigurer extends AsyncConfigurerSupport {  
  
    @Bean  
    public ThreadPoolTaskExecutor asyncExecutor() {  
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();  
        threadPool.setCorePoolSize(3);  
        threadPool.setMaxPoolSize(3);  
        threadPool.setWaitForTasksToCompleteOnShutdown(true);  
        threadPool.setAwaitTerminationSeconds(60 * 15);  
        return threadPool;  
    }  
  
    @Override  
    public Executor getAsyncExecutor() {  
        return asyncExecutor;  
}  

  @Override  
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
    return (ex, method, params) -> ErrorLogger.getInstance().log(String.format("实行异步使命'%s'", method), ex);
}
}

设置自定义的TaskExecutor

因为AsyncConfigurer的默许线程池在源码中为空，Spring经由过程beanFactory.getBean(TaskExecutor.class)先检察是不是有线程池，未设置时，经由过程beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class)，又查询是不是存在默许称号为TaskExecutor的线程池。所以可在项目中，定义称号为TaskExecutor的bean生成一个默许线程池。

 1 @EnableAsync
 2 @Configuration
 3 public class TaskPoolConfig {
 4     @Bean(name = AsyncExecutionAspectSupport.DEFAULT_TASK_EXECUTOR_BEAN_NAME)
 5     public Executor taskExecutor() {
 6         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 7          //中心线程池大小
 8         executor.setCorePoolSize(10);
 9         //最大线程数
10         executor.setMaxPoolSize(20);
11         //行列容量
12         executor.setQueueCapacity(200);
13         //活泼时候
14         executor.setKeepAliveSeconds(60);
15         //线程名字前缀
16         executor.setThreadNamePrefix("taskExecutor-");
17         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
18         return executor;
19     }

      @Bean(name = "new_task")
 5     public Executor taskExecutor() { 6 ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 7 //中心线程池大小 8 executor.setCorePoolSize(10); 9 //最大线程数 10 executor.setMaxPoolSize(20); 11 //行列容量 12 executor.setQueueCapacity(200); 13 //活泼时候 14 executor.setKeepAliveSeconds(60); 15 //线程名字前缀 16 executor.setThreadNamePrefix("taskExecutor-"); 17 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); 18 return executor; 19  }

20 }

多个线程池

@Async注解，运用体系默许或许自定义的线程池（替代默许线程池）。可在项目中设置多个线程池，在异步挪用时，指明须要挪用的线程池称号，如@Async("new_task")。

参考文章

CompletableFuture基础用法

Spring Boot系列二 Spring @Async异步线程池用法总结

Spring Boot@Async默许线程池致使OOM题目

人赞二维码打赏

选择打赏方式

微信QQ钱包支付宝

打赏

打赏

打赏

移步手机端

1、打开你手机的二维码扫描APP
2、扫描左则的二维码
3、点击扫描获得的网址
4、可以在手机端阅读此文章

未定义标签

简介

运用场景

Spring 已完成的异常线程池

Spring中启用@Async

@Async运用默许线程池

@Async运用自定义线程池

完成接口AsyncConfigurer

继续AsyncConfigurerSupport

设置自定义的TaskExecutor

多个线程池

参考文章

发表评论