《SpringBoot2.0 实战》系列-异步多线程调用

比眉伴天荒 2023-08-17 17:07 40阅读 0赞

### 前言： ###

异步在工作中越来越多的被使用到，比如：推送提醒、服务之间调用、数据同步等。最近我就遇到一个需求，【发布公告的时候需要调用第三方的接口发送短信】，这时我们就可以用异步实现短信发送，即可保证接口的响应速度，又可保证接口不受三方接口的超时、异常影响。

> 聊聊异步和同步：
> 
> 同步：一件事一件事的做；【吃饭、睡觉、打豆豆】就是一个同步的过程；  
> 异步：多件事一起做；【边吃边聊】就是一个异步的过程；

举得例子可能不太恰当，只是为了让大家更好的理解。下面我们就进入正题。

![2019123014044767.jpeg][]

### 增加核心配置类： ###

@EnableAsync可加载启动类上，也可加载配置类上

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.core.task.TaskDecorator;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.web.context.request.RequestAttributes;
    import org.springframework.web.context.request.RequestContextHolder;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * 异步线程池配置
     * 
     * @author gourd
     */
    @Configuration
    @EnableAsync
    public class TaskPoolConfig implements AsyncConfigurer {
    
        @Bean("asyncExecutor")
        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            // 核心线程数（默认线程数）
            executor.setCorePoolSize(10);
            // 最大线程数
            executor.setMaxPoolSize(20);
            // 缓冲队列数
            executor.setQueueCapacity(200);
            // 允许线程空闲时间（单位：默认为秒）
            executor.setKeepAliveSeconds(60);
            // 线程池名前缀
            executor.setThreadNamePrefix("taskExecutor-");
            // 设置是否等待计划任务在关闭时完成
            executor.setWaitForTasksToCompleteOnShutdown(true);
            // 设置此执行器应该阻止的最大秒数
            executor.setAwaitTerminationSeconds(60);
            // 增加 TaskDecorator 属性的配置
            executor.setTaskDecorator(new ContextDecorator());
            // 线程池对拒绝任务的处理策略
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            // 初始化
            executor.initialize();
            return executor;
        }
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return null;
        }
    
    
        /**
         * 任务装饰器
         */
        class ContextDecorator implements TaskDecorator {
            @Override
            public Runnable decorate(Runnable runnable) {
                RequestAttributes context = RequestContextHolder.currentRequestAttributes();
                return () -> {
                    try {
                        RequestContextHolder.setRequestAttributes(context);
                        runnable.run();
                    } finally {
                        RequestContextHolder.resetRequestAttributes();
                    }
                };
            }
        }
    }

### service业务逻辑： ###

1.在需要异步的方法上加上  @Async 注解就可以了；

2.如果不想抽出一个异步方法，也可以直接使用线程池，下文测试有demo。

import com.gourd.common.async.service.AsyncService;
    import com.gourd.common.data.BaseResponse;
    import com.gourd.common.utils.RequestHolder;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.scheduling.annotation.AsyncResult;
    import org.springframework.stereotype.Service;
    
    import javax.servlet.http.HttpServletRequest;
    import java.util.concurrent.Future;
    
    /**
     * @author gourd
     */
    @Service
    @Slf4j
    public class AsyncServiceImpl implements AsyncService {
    
        @Override
        @Async
        public Future<BaseResponse> doTaskOne() {
            log.info("开始做任务一（睡眠2s）");
            methodThread();
            HttpServletRequest request = RequestHolder.getRequest();
            log.info("任务一完成，当前线程为 {}，请求方法为 {}，请求路径为：{}", Thread.currentThread().getName(), request.getMethod(), request.getRequestURL().toString());
            return new AsyncResult<>(BaseResponse.ok("任务一完成"));
        }
    
        @Override
        @Async
        public Future<BaseResponse> doTaskTwo(){
            log.info("开始做任务二（睡眠2s）");
            methodThread();
            HttpServletRequest request = RequestHolder.getRequest();
            log.info("任务二完成，当前线程为 {}，请求方法为 {}，请求路径为：{}", Thread.currentThread().getName(), request.getMethod(), request.getRequestURL().toString());
            return new AsyncResult<>(BaseResponse.ok("任务二完成"));
        }
    
        @Override
        @Async
        public void doTaskThree()  {
            log.info("开始做任务三（睡眠1s）");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.interrupted();
                log.error("sleep异常：{}", e);
            }
            HttpServletRequest request = RequestHolder.getRequest();
            log.info("任务三完成，当前线程为 {}，请求方法为 {}，请求路径为：{}", Thread.currentThread().getName(), request.getMethod(), request.getRequestURL().toString());
        }
    
    
        /**
         * 接口睡眠
         */
        private void methodThread() {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                Thread.interrupted();
                log.error("sleep异常：{}", e);
            }
        }
    }

**BaseResponse实体类：**

import lombok.Data;
    import org.springframework.http.HttpStatus;
    
    import java.io.Serializable;
    
    /**
     * @Author: gourd
     * @Description:
     * @Date:Created in 2018/8/27 16:19.
     */
    @Data
    public class BaseResponse<T> implements Serializable {
        private int code;
        private String msg;
        private T data;
        private String token;
        public BaseResponse() {
    
        }
        public BaseResponse(int code, String msg, T data, String token) {
            this.code = code;
            this.msg = msg;
            this.data = data;
            this.token = token;
        }
        public BaseResponse(int code, String msg, T data) {
            this.code = code;
            this.msg = msg;
            this.data = data;
        }
        public BaseResponse(int code, String msg) {
            this.code = code;
            this.msg = msg;
        }
    
    
        public BaseResponse(int code, String msg, String token) {
            this.code = code;
            this.msg = msg;
            this.token = token;
        }
    
        public static BaseResponse ok(Object o) {
            return new BaseResponse(HttpStatus.OK.value(), HttpStatus.OK.getReasonPhrase(),o);
        }
        public static BaseResponse ok(String msg) {
            return new BaseResponse(HttpStatus.OK.value(), msg);
        }
    
        public static BaseResponse ok(String msg,Object o) {
            return new BaseResponse(HttpStatus.OK.value(), msg,o);
        }
        public static BaseResponse failure(String msg) {
            return new BaseResponse(HttpStatus.INTERNAL_SERVER_ERROR.value(), msg);
        }
    
        public static BaseResponse failure( Object o) {
            return new BaseResponse(HttpStatus.BAD_REQUEST.value(), HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase(),o);
        }
    
    
    }

**Controller接口测试**

import com.gourd.common.async.service.AsyncService;
    import com.gourd.common.data.BaseResponse;
    import io.swagger.annotations.Api;
    import io.swagger.annotations.ApiOperation;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.concurrent.Future;
    
    /**
     * @author gourd
     * createAt: 2018/9/17
     */
    
    @RestController
    @Api(tags = "async",description = "异步任务控制器")
    @RequestMapping("/async")
    @Slf4j
    public class AsyncTaskController {
    
        @Autowired
        private AsyncService asyncService;
    
        /**
         * 直接使用异步线程池
         */
        @Autowired
        private Executor asyncExecutor;
    
        @GetMapping(value = "/task" )
        @ApiOperation(value = "异步任务控制器", notes = "异步任务控制器")
        public BaseResponse taskExecute(){
            long startTime = System.currentTimeMillis();
            try {
                Future<BaseResponse> r1 = asyncService.doTaskOne();
                Future<BaseResponse> r2 = asyncService.doTaskTwo();
                asyncService.doTaskThree();
                // 异步线程池执行
                asyncExecutor.execute(() -> log.info("^o^============异步线程池执行...."));
                while (true) {
                    if (r1.isDone() && r2.isDone()) {
                        log.info("异步任务一、二已完成");
                        break;
                    }
                }
                BaseResponse baseResponse1 = r1.get();
                BaseResponse baseResponse2 = r2.get();
                log.info("返回结果：{}，{}",baseResponse1 , baseResponse2);
            } catch (Exception e) {
                log.error("执行异步任务异常 {}",e.getMessage());
            }
            long endTime = System.currentTimeMillis();
            log.info("异步任务总耗时:{}",endTime-startTime);
            return BaseResponse.ok("异步任务全部执行成功");
        }
    }

### 结果分析： ###

点击运行后的打印结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hYTkxZVw_size_16_color_FFFFFF_t_70][]按正常同步的情况，应该是：【任务一】-> 【任务二】-> 【任务三】，且接口总耗时应该是2+2+1 = 5秒左右，但是实际的运行情况只有2秒左右，所以说明了三个任务是异步执行的。任务一和任务二是有返回值得所以用了一个while（true）监控两个任务都执行完后取到返回值，再去处理后续的逻辑。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hYTkxZVw_size_16_color_FFFFFF_t_70 1][] 测试案例流程图

### 结尾： ###

本文是最近实践异步多线程的一些总结和记录，如有不对的地方，欢迎评论吐槽。

===============================================

代码均已上传至本人的开源项目

![20191230131831499.jpeg][]

spring-cloud-plus：[https://blog.csdn.net/HXNLYW/article/details/104635673][https_blog.csdn.net_HXNLYW_article_details_104635673]

[2019123014044767.jpeg]: /images/20230808/d200d804cf1b4a0086cdd2659aa0199c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hYTkxZVw_size_16_color_FFFFFF_t_70]: /images/20230808/ccd3504e0d4c46a0b748f6fa46aa4e2a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hYTkxZVw_size_16_color_FFFFFF_t_70 1]: /images/20230808/2e8a495241a843c6980d070eef7258b8.png
[20191230131831499.jpeg]: /images/20230808/d94b5f03390448caa8844d5ac2400f26.png
[https_blog.csdn.net_HXNLYW_article_details_104635673]: https://blog.csdn.net/HXNLYW/article/details/104635673