异步代码处理

曾经终败给现在 2024-03-30 10:58 72阅读 0赞

在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 CompletableFuture supplyAsync(Supplier supplier)
 public static CompletableFuture supplyAsync(Supplier 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，实现方式如代码所示：  
![在这里插入图片描述][b151369edeb94a8596e3a9daf8212adc.png]

@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在初始化之后会自动调用。需要注意的是，在同级类中直接调用异步方法无法实现异步。

[b151369edeb94a8596e3a9daf8212adc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/cdd2b4e536d9401d87c652a79d02f154.png