Java - 异步处理

缺乏、安全感 2023-09-23 21:19 43阅读 0赞

## 一、异步处理 ##

#### 1、异步概念 ####

异步处理不用阻塞当前线程来等待处理完成，而是允许后续操作，直至其它线程将处理完成，并回调通知此线程。

> 异步是一种设计理念，异步操作不等于多线程，MQ中间件，或者消息广播（这些是可以实现异步处理的方式）

同步处理和异步处理相对，需要实时处理并响应，一旦超过时间会结束会话，在该过程中调用方一直在等待响应方处理完成并返回。同步类似电话沟通，需要实时对话，异步则类似短信交流，发送消息之后无需保持等待状态。

#### 2、异步处理优点 ####

虽然异步处理不能实时响应，但是处理复杂业务场景，多数情况都会使用异步处理。

1、异步可以解耦业务间的流程关联，降低耦合度

2、降低接口响应时间，例如用户注册，异步生成相关信息表；

3、异步可以提高系统性能，提升吞吐量；

4、流量削峰即把请求先承接下来，然后在异步处理；

5、异步用在不同服务间，可以隔离服务，避免雪崩；

异步处理的实现方式有很多种，常见多线程，消息中间件，发布订阅的广播模式，其根据逻辑在于先把请求承接下来，放入容器中，在从容器中把请求取出，统一调度处理。

#### 3、异步处理模式 ####

异步流程处理的实现有好多方式，但是实际开发中常用的就那么几种，例如：

基于接口异步响应，常用在第三方对接流程；

基于消息生产和消费模式，解耦复杂流程；

基于发布和订阅的广播模式，常见系统通知

异步适用的业务场景，对数据强一致性的要求不高，异步处理的数据更多时候追求的是最终一致性

## 二、异步处理4种方式 ##

### **（一）通过创建新线程** ###

异步调用的本质，是通过开启一个新的线程来执行

public class NewThread {
        public static void main(String[] args) {
            System.out.println("主线程 =====> 开始 =====> " + new Date());
    
            new Thread(() -> {
                System.out.println("异步线程 =====> 开始 =====> " + new Date());
                try{
                    Thread.sleep(5000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
                System.out.println("异步线程 =====> 结束 =====> " + new Date());
            }).start();
    
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("主线程 =====> 结束 =====> " + new Date());
        }
    }

### （二）通过线程池 ###

异步任务的实现本质的由新线程来执行任务，所以通过线程池的也可以实现异步执行。写法同我们利用线程池开启多线程一样。但由于我们的目的不是执行多线程，而是异步执行任务，所以一般需要另外一个线程就够了。

因此区别于执行多线程任务的我们常用的newFixedThreadPool，在执行异步任务时，我们用newSingleThreadExecutor 来创建一个单个线程的线程池

public class ThreadPool {
        public static void main(String[] args) {
            System.out.println("主线程 =====> 开始 =====> " + new Date());
    
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            executorService.submit(()->{
                System.out.println("异步线程 =====> 开始 =====> " + new Date());
                try{
                    Thread.sleep(5000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
                System.out.println("异步线程 =====> 结束 =====> " + new Date());
            });
            executorService.shutdown(); // 回收线程池
    
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("主线程 =====> 结束 =====> " + new Date());
        }
    }

### （三）通过SpringBoot中@Async注解 ###

SpringBoot项目有一个的很重要的特点就是的注解化。项目是SpringBoot，就可选择使用[@Async][Async]注解

将要异步执行的代码封装成一个方法，然后用@Async注解该方法，然后在主方法中直接调用就行

public class UseAsyncAnnotation {
        public static void main(String[] args) {
            System.out.println("主线程 =====> 开始 =====> " + new Date());
    
            AsyncThread asyncThread = new AsyncThread();
            asyncThread.asyncThread();
            
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            System.out.println("主线程 =====> 结束 =====> " + new Date());
    
            try {
                Thread.sleep(4000); // 用于防止jvm停止，导致异步线程中断
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    public class AsyncThread{
        @Async
        public void asyncThread(){
            System.out.println("异步线程 =====> 开始 =====> " + new Date());
            try{
                Thread.sleep(5000);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
            System.out.println("异步线程 =====> 结束 =====> " + new Date());
        }
    }

#### 使用@Async注解注意事项 ####

> **1、主启动类需要添加@EnableAsync注解**  
> **2、@Async注解的方法与调用方法不能在同一个类中，否则不生效**  
> **@Async注解是基础Spring的AOP动态代理模式实现的，同类中的A方法并没有走Spring的代理类，调用同类中的B方法是原生的并不是Spring所代理的对象，因为并没有经过Spring容器托管，从而造成了注解的失效**  
> **3、需要通过@Autowired或@Resource进行注入，不可手动new**  
> **4、被@Async注解修饰的方法必须不可以是static和private，必须为public**

@RestController
    public class UserController {
     
        @Autowired
        private UserService userService;
     
        @RequestMapping("/getAll")
        public void getUsers(){
            System.out.println("业务开始");
            userService.getUsers();
            System.out.println("业务结束");
        }
     
    }
    
    public interface UserService {
        @Async
        List<User> getUsers();
    }

可以通过添加配置项处理**@EnableAsync注解**

@Configuration
    @EnableAsync
    @ComponentScan("com.service")
    public class SpringAsyncConfigure {
        /** Set the ThreadPoolExecutor's core pool size. */
        private int corePoolSize = 10;
        /** Set the ThreadPoolExecutor's maximum pool size. */
        private int maxPoolSize = 200;
        /** Set the capacity for the ThreadPoolExecutor's BlockingQueue. */
        private int queueCapacity = 10;
    
        private String ThreadNamePrefix = "executor-";
    
        @Bean
        public Executor estoreExecutor(){
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(corePoolSize);
            executor.setMaxPoolSize(maxPoolSize);
            executor.setQueueCapacity(queueCapacity);
            executor.setThreadNamePrefix(ThreadNamePrefix);
    
            // rejection-policy：当pool已经达到max size的时候，如何处理新任务
            // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executor.initialize();
            return executor;
        }
    
    }

在 xxServiceImpl中方法添加注解 **@Async**

@Service
    public class xxServiceImpl extends BaseService implements xxService {
        @Override
        @Async(value = "xxExecutor")
        public void syncPicSave() {
            do somthing......
        }
    }

### （四）通过CompletableFuture ###

CompletableFuture是JDK1.8的新特性，是对Future的扩展。CompletableFuture实现了CompletionStage接口和Future接口，增加了异步回调、流式处理、多个Future组合处理的能力

public class CompletableFutureTest {
        public static void main(String[] args) {
            System.out.println("主线程 =====> 开始 =====> " + new Date());
    
            ExecutorService executorService = Executors.newSingleThreadExecutor();
    
            CompletableFuture.runAsync(() ->{
                System.out.println("异步线程 =====> 开始 =====> " + new Date());
                try{
                    Thread.sleep(5000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
                System.out.println("异步线程 =====> 结束 =====> " + new Date());
            },executorService);
    
            executorService.shutdown(); // 回收线程池
    
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            System.out.println("主线程 =====> 结束 =====> " + new Date());
        }
    }

[Async]: mailto:%E2%80%94%E2%80%94@Async