Async 使用详解

蔚落 2023-10-12 09:00 70阅读 0赞

# Spring Boot异步调用@Async #

在实际开发中，有时候为了及时处理请求和进行响应，我们可能会多任务同时执行，或者先处理主任务，也就是异步调用，异步调用的实现有很多，例如多线程、定时任务、消息队列等，

# 一、普通串行执行演示 #

## 1.1任务类 ##

假设有三个任务需要处理我们在平时开发中，会按照逻辑顺序依次编写代码；

@Component
    public class Task {
        
    
        public static Random random =new Random();
    
        public void doTaskOne() throws Exception {
        
            System.out.println("开始做任务一");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务一，耗时：" + (end - start) + "毫秒");
        }
    
        public void doTaskTwo() throws Exception {
        
            System.out.println("开始做任务二");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(5000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务二，耗时：" + (end - start) + "毫秒");
        }
    
        public void doTaskThree() throws Exception {
        
            System.out.println("开始做任务三");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务三，耗时：" + (end - start) + "毫秒");
        }
    }

## 1.2 测试类 ##

@SpringBootTest
    class ApplicationTests {
        
    
        @Test
        void contextLoads() {
        
        }
        @Autowired
        private Task task;
    
        @Test
        public void test() throws Exception {
        
            task.doTaskOne();
            task.doTaskTwo();
            task.doTaskThree();
        }
    }

## 1.3运行结果及分析 ##

观察结果发现，先调用的任务一，任务一执行完成才能开始任务二，任务二完成才能执行任务三；这种串行话的执行程序在实际开发中并没以后任务问题  
![在这里插入图片描述][35bd574f16c24390a5adf8ddfda38ef7.png]  
假如有这样一个功能，任务二是发送邮件，任务三是发送短信，实际开发中这两个任务互补干扰；谁先谁后都都可以，如果是串行操作是不是有点效率低了！最好的办法可以让这两个任务并行操作  
**实现异步的方式：**

*  1、使用多线程，在主线程中分别给任务一，任务二另起一个线程来执行任务
 *  2、使用@Async注解实现异步

# 二、@Async使用演示 #

## 2.1 在启动类上添加@EnableAsync开启异步 ##

`@Async`是`Spring内置注解`，用来处理异步任务，在`SpringBoot`中同样适用，且在`SpringBoot`项目中，除了boot本身的starter外，不需要额外引入依赖。  
而要使用`@Async`，需要在 **启动类上加上`@EnableAsync`主动声明来开启异步方法。**

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

## 2.2 编写任务类，在任务方法添加@Async ##

@Component
    public class TaskAsync {
        
        public static Random random = new Random();
    
        @Async
        public void doTaskOne() throws Exception {
        
            System.out.println("开始做任务一");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务一，耗时：" + (end - start) + "毫秒");
        }
    
        @Async
        public void doTaskTwo() throws Exception {
        
            System.out.println("开始做任务二");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务二，耗时：" + (end - start) + "毫秒");
        }
    
        @Async
        public void doTaskThree() throws Exception {
        
            System.out.println("开始做任务三");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务三，耗时：" + (end - start) + "毫秒");
        }
    }

2.3测试类

@SpringBootTest
    class ApplicationTests {
        
    
        @Test
        void contextLoads() {
        
        }
        @Autowired
        private Task task;
        @Autowired
        private TaskAsync taskAsync;
    
        @Test
        public void test() throws Exception {
        
            task.doTaskOne();
            task.doTaskTwo();
            task.doTaskThree();
        }
    
        @Test
        public void taskAsynctest() throws Exception {
        
            try {
        
                long start = System.currentTimeMillis();
                taskAsync.doTaskOne();
                taskAsync.doTaskTwo();
                taskAsync.doTaskThree();
                Thread.sleep(10000);
                long end = System.currentTimeMillis();
                System.out.println("end = " + (end - start)/1000f);
    
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            
        }
    }

## 2.3 多次运行观察结果 ##

然后我们在运行测试类，这个时候输出可能就五花八门了，任意任务都可能先执行完成，也有可能有的方法因为主程序关闭而没有输出。  
![在这里插入图片描述][1abb42eaa9f84be7bc2fba1708835d47.png]

## 2.4总结 ##

使用 @Async可以实现程序的异步执行，完成程序优化；但实际使用中需要注意异步失效问题：  
**被调用方法和调用方法处理同一个类中会导致异步失效**

*  同一个类中。 失效的代码

class TestService {
        
        void a() {
         
          this.b();
        }
    	
        @Async
        void b(){
        }
    }

*  正常的代码

class TestService {
        
        void a(){
         
           BService.b();
        }
    }
    
    class BService() {
        
        @Async
        void b(){
        }
    }

[从@Async案例找到Spring框架的bug：exposeProxy=true不生效原因大剖析+最佳解决方案][Async_Spring_bug_exposeProxy_true]  
[@Async失效情况][Async]

# 三、@Async + 线程池使用 #

@Async 异步方法默认使用 Spring 创建 ThreadPoolTaskExecutor (参考TaskExecutionAutoConfiguration)，其中默认核心线程数为 8，默认最大队列和默认最大线程数都是 Integer.MAX\_VALUE。创建新线程的条件是队列填满时，而这样的配置队列永远不会填满，如果有 @Async 注解标注的方法长期占用线程 (比如 HTTP 长连接等待获取结果)，在核心 8 个线程数占用满了之后，新的调用就会进入队列，外部表现为没有执行。

**我们可以自定义一个线程池，线程数的设定需要考虑一下要执行的任务是 IO 密集型任务，还是 CPU 密集型任务。对于 CPU 密集型任务，如 CPU 核数 + 1；对于 IO 密集型任务，由于 IO 密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如 CPU 核数 \* 2。**

## 3.1线程池配置类 ##

接下来给出一个 IO 密集型任务的线程池配置代码

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.ThreadPoolExecutor;
    
    @Configuration
    public class ThreadPoolConfig {
        
    
        /**
         * 核心线程数
         */
        private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
    
        /**
         * 最大线程数
         */
        private static final int MAX_POOL_SIZE = CORE_POOL_SIZE * 4 < 256 ? 256 : CORE_POOL_SIZE * 4;
    
        /**
         * 允许线程空闲时间（单位为秒）
         */
        private static final int KEEP_ALIVE_TIME = 10;
    
        /**
         * 缓冲队列数
         */
        private static final int QUEUE_CAPACITY = 200;
    
        /**
         * 线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁
         */
        private static final int AWAIT_TERMINATION = 60;
    
        /**
         * 用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
         */
        private static final Boolean WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN = true;
    
        /**
         * 线程池名前缀
         */
        private static final String THREAD_NAME_PREFIX = "Spider-ThreadPool-";
    
    
        @Bean("spiderTaskExecutor")
        public ThreadPoolTaskExecutor spiderTaskExecutor () {
        
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
            taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
            taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
            taskExecutor.setQueueCapacity(QUEUE_CAPACITY);
            taskExecutor.setThreadNamePrefix(THREAD_NAME_PREFIX);
            taskExecutor.setWaitForTasksToCompleteOnShutdown(WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN);
            taskExecutor.setAwaitTerminationSeconds(AWAIT_TERMINATION);
            /**
             * 拒绝策略 => 当pool已经达到max size的时候，如何处理新任务
             * CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
             * AbortPolicy：直接抛出异常，这是默认策略；
             * DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
             * DiscardPolicy：直接丢弃任务；
             */
            taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            taskExecutor.initialize();
            return taskExecutor;
        }
    
    }

## 3.2 修改任务类 ##

在任务类中的方法上添加`@Async("spiderTaskExecutor")`

@Component
    public class TaskAsync {
        
        public static Random random = new Random();
    
        @Async("spiderTaskExecutor")
        public void doTaskOne() throws Exception {
        
            System.out.println("开始做任务一");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务一，耗时：" + (end - start) + "毫秒");
        }
    
        @Async("spiderTaskExecutor")
        public void doTaskTwo() throws Exception {
        
            System.out.println("开始做任务二");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务二，耗时：" + (end - start) + "毫秒");
        }
    
        @Async("spiderTaskExecutor")
        public void doTaskThree() throws Exception {
        
            System.out.println("开始做任务三");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            System.out.println("完成任务三，耗时：" + (end - start) + "毫秒");
        }
    }

最后补充一些知识，要合理的控制线程数（比如采集订单信息的同时要采集订单详情和文章信息，订单详情和文章信息可以合并在一个线程中处理），不要滥用。需要考虑什么时候使用 MQ，什么时候开启线程异步处理。推荐一个分析 jstack 文件的工具，IBM Thread and Monitor Dump Analyzer for Java，分析一下正在运行、发生死锁、等待、阻塞的线程。

注意：[@async失效原因][async]

# 四 获取异步执行结果 #

上面演示了@Async，但是有时候除了需要任务并发调度外，我们还需要获取任务的返回值，且在多任务都执行完成后再结束主任务，这个时候又该怎么处理呢？  
[CompletableFuture使用详解][CompletableFuture]

# 五源码地址 #

[源码传送带][Link 1]

# 六扩展思考 #

有没有想过如果异步调用失败了，我们该怎么保证调用的正确性？  
在实际工作中，重处理是一个非常常见的场景，比如:

*  发送消息失败
 *  调用远程服务失败。
 *  争抢锁失败。

这些错误可能是因为网络波动造成的，等待过后重处理就能成功。通常来说，会用try/catch，while循环之类的语法来进行重处理，但是这样的做法缺乏统一性，并且不是很方便，要多写很多代码。然而spring-retry却可以通过注解，在不入侵原有业务逻辑代码的方式下，优雅的实现重处理功能。

[@Retryable（spring的重试机制）][Retryable_spring]

[35bd574f16c24390a5adf8ddfda38ef7.png]: https://img-blog.csdnimg.cn/35bd574f16c24390a5adf8ddfda38ef7.png
[1abb42eaa9f84be7bc2fba1708835d47.png]: https://img-blog.csdnimg.cn/1abb42eaa9f84be7bc2fba1708835d47.png
[Async_Spring_bug_exposeProxy_true]: https://cloud.tencent.com/developer/article/1497700
[Async]: https://blog.csdn.net/ElendaLee/article/details/127411225
[async]: https://juejin.cn/post/6844903839800229895
[CompletableFuture]: https://blog.csdn.net/weixin_43811057/article/details/119577690
[Link 1]: https://gitee.com/wl8888/async.git
[Retryable_spring]: https://blog.csdn.net/weixin_43811057/article/details/131759900?spm=1001.2014.3001.5501