@Async异步操作及异步线程池

本文为博主原创，转载请注明出处：

　　@Async 用来实现异步请求操作，使用@Async 注解时，需要同时使用 @EnableAsync 注解，使用 @EnableAsync 注解用于开启异步请求。

　　如果没有使用 @EnableAsync 注解，则不会开启异步操作，是同步请求。

　　一。异步方法调用及讲解

　　使用场景： 当不影响当前主线程的功能，新建线程进行其他功能。比如：异步消息通知，或异步文件上传等等。

　　封装一个使用 @Async 的service 类，进行测试调用：

package com.example.demo.service.impl;

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

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;

@Service
@EnableAsync
public class AsyncServiceImpl {

    // 异步调用无返回
    @Async
    public void getAsyncNum() throws InterruptedException {
        System.out.println("==AsyncServiceImpl=getAsyncNum==="+Thread.currentThread().getName());
        Thread.sleep(3000);
    }

    // 没有使用 @Async 注解，则为同步方法
    public Integer getSyncNum(){
        System.out.println("==AsyncServiceImpl=getSyncNum==="+Thread.currentThread().getName());
        Random random = new Random();
        int num = random.nextInt(333333);
        return num;
    }
    
    // 异步调用返回一个Future 类型的参数
    @Async
    public Future<String> getAsyncFuture() throws InterruptedException {
        int thinking = 2;
        Thread.sleep(thinking * 1000);
        System.out.println("getAsyncFuture==="+Thread.currentThread().getName());
        return new AsyncResult<String>("发送消息用了"+thinking+"秒");
    }

    // 异步调用 返回一个 CompletableFuture 类型的参数
    @Async
    public CompletableFuture<Integer> completableFutureTask() {
        System.out.println("---completableFutureTask---"+Thread.currentThread().getName());
        Random random = new Random();
        int num = random.nextInt(333333);
        // 模拟这是一个耗时的任务
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //返回一个已经用给定值完成的新的CompletableFuture。
        return CompletableFuture.completedFuture(num);
    }
}

　1. 调用一个异步无返回参数的方法

　　 调用一个异步无返回参数的方法，则直接调用既可，异步方法会新开启一个线程，执行异步方法：

 @Test
    void queryAsync() throws InterruptedException {
        System.out.println("----queryAsync--"+Thread.currentThread().getName());
        asyncService.getAsyncNum();
        System.out.println("==queryAsync=end=");
    }

　　打印的描述为：

  可以发现，测试方法会立即结束，异步方法还在执行，相互不影响。

　　2. 调用同步方法，查看线程：

 @Test
    void test2(){
        System.out.println("---test2---"+Thread.currentThread().getName());
        int num = asyncService.getSyncNum();
    }

打印结果为：

 　　没有使用 @Async 注解的方法，直接调用，整个方法都是同一线程。

　　3. 将 上面同步方法不用 @Async 改进为 异步方法，

　　使用 FutureTask 类封装异步方法，且需新建并启动一个新的Thread，实现异步功能：

 @Test
    void test3() throws ExecutionException, InterruptedException {
        System.out.println("---test3---"+Thread.currentThread().getName());
        FutureTask<Integer> task = new FutureTask<>(()->asyncService.getSyncNum());
        Thread thread = new Thread(task);
        thread.start();
        int num = task.get()+ 3;
        System.out.println(num);
    }

       打印的日志为：

　　通过 FutureTask 与Thread 实现一个异步方法功能，发现方法调用与当前方法是两个线程。

　　4.  @Async 注解进行异步方法及存在返回值使用 AsyncResult 封装

　　当异步方法存在返回参数时，必须使用Future 类或相关子类进行封装，如果不进行封装，则调用异步方法时，会直接返回null，在解析或使用返回参数时，就会报空指针异常。　　

　　使用 @Async 注解进行异步方法，存在返回使，可以使用 AsyncResult 类进行封装。其也是 future 的子类。可以看 getAsyncFuture 中的使用和返回。

@Test
    void test4() throws InterruptedException, ExecutionException {
        System.out.println("---test4---"+Thread.currentThread().getName());
        Future<String> result = asyncService.getAsyncFuture();
        String result2 = result.get();
        System.out.println(result2);
    }

　　上述测试方法执行打印的日志如下：

            

       调用 getAsyncFuture（）异步方法，会新建一个线程执行，且该方法为有参数返回。使用 AsyncResult 与Future 相关封装。获取返回值时使用 get（）方法。

该方法会等待异步方法返回结果。

　　5. 使用 CompletableFuture 类封装异步返回结果。

　　该类也是 Future 的子类。其使用方法可参考最上面  completableFutureTask（）方法。

@Test
    void test5() throws ExecutionException, InterruptedException {
        System.out.println("---test5---"+Thread.currentThread().getName());
        CompletableFuture<Integer> result = asyncService.completableFutureTask();
        System.out.println(result.get());
    }

     进行测试方法调用：

            

        CompletableFuture 与 AsyncResult  使用场景相同，都是使用get（）获取返回异步方法返回结果，且会等待异步方法执行结束。

　　二。 使用 ThreadPoolTaskExecutor 定义异步线程池

package com.example.demo.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
public class AsyncConfig {
    private static final int CORE_POOL_SIZE = 6;
    private static final int MAX_POOL_SIZE = 10;
    private static final int QUEUE_CAPACITY = 100;

    @Bean
    public Executor taskExecutor() {
        // Spring 默认配置是核心线程数大小为1，最大线程容量大小不受限制，队列容量也不受限制。
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数
        executor.setCorePoolSize(CORE_POOL_SIZE);
        // 最大线程数
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        // 队列大小
        executor.setQueueCapacity(QUEUE_CAPACITY);
        // 当最大池已满时，此策略保证不会丢失任务请求，但是可能会影响应用程序整体性能。
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setThreadNamePrefix("My ThreadPoolTaskExecutor-");
        executor.initialize();
        return executor;
    }
}

　　

　　ThreadPoolTaskExecutor 常见概念：

• Core Pool Size : 核心线程数线程数定义了最小可以同时运行的线程数量。
• Queue Capacity : 当新任务来的时候会先判断当前运行的线程数量是否达到核心线程数，如果达到的话，信任就会被存放在队列中。
• Maximum Pool Size : 当队列中存放的任务达到队列容量的时候，当前可以同时运行的线程数量变为最大线程数。

　　 一般情况下不会将队列大小设为：Integer.MAX_VALUE,也不会将核心线程数和最大线程数设为同样的大小，这样的话最大线程数的设置都没什么意义了，

你也无法确定当前 CPU 和内存利用率具体情况如何。

　　

　　如果队列已满并且当前同时运行的线程数达到最大线程数的时候，如果再有新任务过来会发生什么呢？

　　Spring 默认使用的是 ThreadPoolExecutor.AbortPolicy。在Spring的默认情况下，ThreadPoolExecutor 将抛出 RejectedExecutionException 来拒绝新来的任务 ，

这代表你将丢失对这个任务的处理。 对于可伸缩的应用程序，建议使用 ThreadPoolExecutor.CallerRunsPolicy。当最大池被填满时，此策略为我们提供可伸缩队列。

　　ThreadPoolTaskExecutor 饱和策略定义:

如果当前同时运行的线程数量达到最大线程数量时，ThreadPoolTaskExecutor 定义一些策略:

• ThreadPoolExecutor.AbortPolicy：抛出 RejectedExecutionException来拒绝新任务的处理。
• ThreadPoolExecutor.CallerRunsPolicy：调用执行自己的线程运行任务。您不会任务请求。但是这种策略会降低对于新任务提交速度，影响程序的整体性能。另外，这个策略喜欢增加队列容量。如果您的应用程序可以承受此延迟并且你不能任务丢弃任何一个任务请求的话，你可以选择这个策略。
• ThreadPoolExecutor.DiscardPolicy： 不处理新任务，直接丢弃掉。
• ThreadPoolExecutor.DiscardOldestPolicy： 此策略将丢弃最早的未处理的任务请求。

　　测试方法调用异步线程池异步方法：

  @Test
    void test6() throws InterruptedException {
        System.out.println("----test6--"+Thread.currentThread().getName());
        for (int i =0;i<6;i++){
            asyncService.getAsyncNum();
        }
        System.out.println("==test6=end=");
    }

　　调用打印的异常日志如下：