乐趣区

关于后端:都在建议你不要直接使用-Async-注解为什么

jiezi

本文讲述 @Async 注解，在 Spring 体系中的利用。本文仅阐明 @Async 注解的利用规定，对于原理，调用逻辑，源码剖析，暂不介绍。对于异步办法调用，从 Spring3 开始提供了 @Async 注解，该注解能够被标注在办法上，以便异步地调用该办法。调用者将在调用时立刻返回，办法的理论执行将提交给 Spring TaskExecutor 的工作中，由指定的线程池中的线程执行。

在我的项目利用中，@Async 调用线程池，举荐应用自定义线程池的模式。自定义线程池罕用计划：从新实现接口 AsyncConfigurer。

| 简介

利用场景

同步：同步就是整个处理过程程序执行，当各个过程都执行结束，并返回后果。

异步：异步调用则是只是发送了调用的指令，调用者无需期待被调用的办法齐全执行结束；而是继续执行上面的流程。

例如，在某个调用中，须要顺序调用 A, B, C 三个过程办法；如他们都是同步调用，则须要将他们都程序执行结束之后，方算作过程执行结束；如 B 为一个异步的调用办法，则在执行完 A 之后，调用 B，并不期待 B 实现，而是执行开始调用 C，待 C 执行结束之后，就意味着这个过程执行结束了。在 Java 中，个别在解决相似的场景之时，都是基于创立独立的线程去实现相应的异步调用逻辑，通过主线程和不同的业务子线程之间的执行流程，从而在启动独立的线程之后，主线程继续执行而不会产生停滞期待的状况。

Spring 曾经实现的线程池

  • SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，默认每次调用都会创立一个新的线程。
  • SyncTaskExecutor：这个类没有实现异步调用，只是一个同步操作。只实用于不须要多线程的中央。
  • ConcurrentTaskExecutor：Executor 的适配类，不举荐应用。如果 ThreadPoolTaskExecutor 不满足要求时，才用思考应用这个类。
  • SimpleThreadPoolTaskExecutor：是 Quartz 的 SimpleThreadPool 的类。线程池同时被 quartz 和非 quartz 应用，才须要应用此类。
  • ThreadPoolTaskExecutor：最常应用，举荐。其实质是对 java.util.concurrent.ThreadPoolExecutor 的包装。

异步的办法有：

  • 最简略的异步调用，返回值为 void。
  • 带参数的异步调用，异步办法能够传入参数。
  • 存在返回值，常调用返回 Future。

| Spring 中启用 @Async

// 基于 Java 配置的启用形式：@Configuration
@EnableAsync
public class SpringAsyncConfig {...}

// Spring boot 启用：@EnableAsync
@EnableTransactionManagement
public class SettlementApplication {public static void main(String[] args) {SpringApplication.run(SettlementApplication.class, args);
    }
}

| @Async 利用默认线程池

Spring 利用默认的线程池，指在 @Async 注解在应用时，不指定线程池的名称。查看源码，@Async 的默认线程池为 SimpleAsyncTaskExecutor。

无返回值调用 

 基于 @Async 无返回值调用，间接在应用类，应用办法（倡议在应用办法）上，加上注解。若须要抛出异样，需手动 new 一个异样抛出。/**
     * 带参数的异步调用 异步办法能够传入参数
     *  对于返回值是 void，异样会被 AsyncUncaughtExceptionHandler 解决掉
     * @param s
     */
    @Async
    public void asyncInvokeWithException(String s) {log.info("asyncInvokeWithParameter, parementer={}", s);
        throw new IllegalArgumentException(s);
    }

有返回值 Future 调用 

/**
     * 异样调用返回 Future
     *  对于返回值是 Future，不会被 AsyncUncaughtExceptionHandler 解决，须要咱们在办法中捕捉异样并解决
     *  或者在调用方在调用 Futrue.get 时捕捉异样进行解决
     *
     * @param i
     * @return
     */
    @Async
    public Future<String> asyncInvokeReturnFuture(int i) {log.info("asyncInvokeReturnFuture, parementer={}", i);
        Future<String> future;
        try {Thread.sleep(1000 * 1);
            future = new AsyncResult<String>("success:" + i);
            throw new IllegalArgumentException("a");
        } catch (InterruptedException e) {future = new AsyncResult<String>("error");
        } catch(IllegalArgumentException e){future = new AsyncResult<String>("error-IllegalArgumentException");
        }
        return future;
    }

有返回值 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 接口。

    /**
   * 数据查问线程池
   */
  private static final ThreadPoolExecutor SELECT_POOL_EXECUTOR = new ThreadPoolExecutor(10, 20, 5000,
          TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), new ThreadFactoryBuilder().setNameFormat("selectThreadPoolExecutor-%d").build());

// tradeMapper.countTradeLog(tradeSearchBean) 办法示意，获取数量，返回值为 int
 // 获取总条数
      CompletableFuture<Integer> countFuture = CompletableFuture
              .supplyAsync(() -> tradeMapper.countTradeLog(tradeSearchBean), SELECT_POOL_EXECUTOR);
// 同步阻塞
  CompletableFuture.allOf(countFuture).join();
// 获取后果
 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）。自定义线程池有如下模式：

  • 从新实现接口 AsyncConfigurer；
  • 继承 AsyncConfigurerSupport；
  • 配置由自定义的 TaskExecutor 代替内置的工作执行器。

通过查看 Spring 源码对于 @Async 的默认调用规定，会优先查问源码中实现 AsyncConfigurer 这个接口的类，实现这个接口的类为 AsyncConfigurerSupport。但默认配置的线程池和异步解决办法均为空，所以，无论是继承或者从新实现接口，都需指定一个线程池。且从新实现 public Executor getAsyncExecutor() 办法。

实现接口 AsyncConfigurer

@Configuration
 public class AsyncConfiguration implements AsyncConfigurer {@Bean("kingAsyncExecutor")
     public ThreadPoolTaskExecutor executor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
         int corePoolSize = 10;
         executor.setCorePoolSize(corePoolSize);
         int maxPoolSize = 50;
         executor.setMaxPoolSize(maxPoolSize);
         int queueCapacity = 10;
         executor.setQueueCapacity(queueCapacity);
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         String threadNamePrefix = "kingDeeAsyncExecutor-";
         executor.setThreadNamePrefix(threadNamePrefix);
         executor.setWaitForTasksToCompleteOnShutdown(true);
         // 应用自定义的跨线程的申请级别线程工厂类 19         int awaitTerminationSeconds = 5;
         executor.setAwaitTerminationSeconds(awaitTerminationSeconds);
         executor.initialize();
         return executor;
     }
 
     @Override
     public Executor getAsyncExecutor() {return executor();
     }
 
     @Override
     public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {return (ex, method, params) -> ErrorLogger.getInstance().log(String.format("执行异步工作'%s'", method), ex);
     }
 }

继承 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 生成一个默认线程池。也可不指定线程池的名称，申明一个线程池，自身底层是基于 TaskExecutor.class 便可。比方：
Executor.class:ThreadPoolExecutorAdapter->ThreadPoolExecutor->AbstractExecutorService->ExecutorService->Executor

这样的模式，最终底层为 Executor.class，在替换默认的线程池时，需设置默认的线程池名称为 TaskExecutor。

TaskExecutor.class:ThreadPoolTaskExecutor->SchedulingTaskExecutor->AsyncTaskExecutor->TaskExecutor
这样的模式，最终底层为 TaskExecutor.class，在替换默认的线程池时，可不指定线程池名称。

@EnableAsync
 @Configuration
 public class TaskPoolConfig {@Bean(name = AsyncExecutionAspectSupport.DEFAULT_TASK_EXECUTOR_BEAN_NAME)
     public Executor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
          // 外围线程池大小
         executor.setCorePoolSize(10);
         // 最大线程数
         executor.setMaxPoolSize(20);
         // 队列容量
         executor.setQueueCapacity(200);
         // 沉闷工夫
         executor.setKeepAliveSeconds(60);
         // 线程名字前缀
         executor.setThreadNamePrefix("taskExecutor-");
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         return executor;
     }
    @Bean(name = "new_task")
     public Executor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
          // 外围线程池大小
         executor.setCorePoolSize(10);
         // 最大线程数
         executor.setMaxPoolSize(20);
         // 队列容量
         executor.setQueueCapacity(200);
         // 沉闷工夫
         executor.setKeepAliveSeconds(60);
         // 线程名字前缀
         executor.setThreadNamePrefix("taskExecutor-");
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         return executor;
     }
 }

多个线程池

@Async 注解，应用零碎默认或者自定义的线程池（代替默认线程池）。可在我的项目中设置多个线程池，在异步调用时，指明须要调用的线程池名称，如 @Async(“new_task”)。

| @Async 局部重要源码解析

源码 - 获取线程池办法

源码 - 设置默认线程池 defaultExecutor，默认是空的，当从新实现接口 AsyncConfigurer 的 getAsyncExecutor() 时，能够设置默认的线程池。

源码 - 寻找零碎默认线程池

源码 - 都没有找到我的项目中设置的默认线程池时，采纳 spring 默认的线程池

起源：https://www.cnblogs.com/wland…

退出移动版