您当前的位置:首页 > 电脑百科 > 程序开发 > 框架

SpringBoot异步线程,父子线程数据传递的5种姿势

时间:2022-09-16 15:33:09  来源:今日头条  作者:小心程序猿QAQ

姿势1:ThreadLocal+TaskDecorator

用户工具类

/**
 *使用ThreadLocal存储共享的数据变量,如登录的用户信息
 */
public class UserUtils {
    private static  final  ThreadLocal<String> userLocal=new ThreadLocal<>();

    public static  String getUserId(){
        return userLocal.get();
    }
    public static void setUserId(String userId){
        userLocal.set(userId);
    }

    public static void clear(){
        userLocal.remove();
    }

}
复制代码

自定义CustomTaskDecorator

/**
 * 线程池修饰类
 */
public class CustomTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        // 获取主线程中的请求信息(我们的用户信息也放在里面)
        String robotId = UserUtils.getUserId();
        System.out.println(robotId);
        return () -> {
            try {
                // 将主线程的请求信息,设置到子线程中
                UserUtils.setUserId(robotId);
                // 执行子线程,这一步不要忘了
                runnable.run();
            } finally {
                // 线程结束,清空这些信息,否则可能造成内存泄漏
                UserUtils.clear();
            }
        };
    }
}
复制代码
ExecutorConfig

在原来的基础上增加 executor.setTaskDecorator(new CustomTaskDecorator());

@Bean(name = "asyncServiceExecutor")
    public Executor asyncServiceExecutor() {
        log.info("start asyncServiceExecutor----------------");
        //ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //使用可视化运行状态的线程池
        ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(corePoolSize);
        //配置最大线程数
        executor.setMaxPoolSize(maxPoolSize);
        //配置队列大小
        executor.setQueueCapacity(queueCapacity);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix(namePrefix);

        // rejection-policy:当pool已经达到max size的时候,如何处理新任务
        // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        //增加线程池修饰类
        executor.setTaskDecorator(new CustomTaskDecorator());
        //增加MDC的线程池修饰类
        //executor.setTaskDecorator(new MDCTaskDecorator());
        //执行初始化
        executor.initialize();
        log.info("end asyncServiceExecutor------------");
        return executor;
    }
复制代码
AsyncServiceImpl
/**
     * 使用ThreadLocal方式传递
     * 带有返回值
     * @throws InterruptedException
     */
    @Async("asyncServiceExecutor")
    public CompletableFuture<String> executeValueAsync2() throws InterruptedException {
        log.info("start executeValueAsync");
        System.out.println("异步线程执行返回结果......+");
        log.info("end executeValueAsync");
        return CompletableFuture.completedFuture(UserUtils.getUserId());
    }

复制代码
Test2Controller
/**
     * 使用ThreadLocal+TaskDecorator的方式
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @GetMApping("/test2")
    public String test2() throws InterruptedException, ExecutionException {
        UserUtils.setUserId("123456");
        CompletableFuture<String> completableFuture = asyncService.executeValueAsync2();
        String s = completableFuture.get();
        return s;
    }
复制代码

姿势2:RequestContextHolder+TaskDecorator

自定义CustomTaskDecorator

/**
 * 线程池修饰类
 */
public class CustomTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        // 获取主线程中的请求信息(我们的用户信息也放在里面)
        RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
        return () -> {
            try {
                // 将主线程的请求信息,设置到子线程中
                RequestContextHolder.setRequestAttributes(attributes);
                // 执行子线程,这一步不要忘了
                runnable.run();
            } finally {
                // 线程结束,清空这些信息,否则可能造成内存泄漏
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
}
复制代码
ExecutorConfig

在原来的基础上增加 executor.setTaskDecorator(new CustomTaskDecorator());

@Bean(name = "asyncServiceExecutor")
    public Executor asyncServiceExecutor() {
        log.info("start asyncServiceExecutor----------------");
        //ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //使用可视化运行状态的线程池
        ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(corePoolSize);
        //配置最大线程数
        executor.setMaxPoolSize(maxPoolSize);
        //配置队列大小
        executor.setQueueCapacity(queueCapacity);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix(namePrefix);

        // rejection-policy:当pool已经达到max size的时候,如何处理新任务
        // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        //增加线程池修饰类
        executor.setTaskDecorator(new CustomTaskDecorator());
        //增加MDC的线程池修饰类
        //executor.setTaskDecorator(new MDCTaskDecorator());
        //执行初始化
        executor.initialize();
        log.info("end asyncServiceExecutor------------");
        return executor;
    }
复制代码
AsyncServiceImpl
/**
     * 使用RequestAttributes获取主线程传递的数据
     * @return
     * @throws InterruptedException
     */
    @Async("asyncServiceExecutor")
    public CompletableFuture<String> executeValueAsync3() throws InterruptedException {
        log.info("start executeValueAsync");
        System.out.println("异步线程执行返回结果......+");
        RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
        Object userId = attributes.getAttribute("userId", 0);
        log.info("end executeValueAsync");
        return CompletableFuture.completedFuture(userId.toString());
    }

复制代码
Test2Controller
/**
     * RequestContextHolder+TaskDecorator的方式
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @GetMapping("/test3")
    public String test3() throws InterruptedException, ExecutionException {
        RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
        attributes.setAttribute("userId","123456",0);
        CompletableFuture<String> completableFuture = asyncService.executeValueAsync3();
        String s = completableFuture.get();
        return s;
    }
复制代码

姿势3:MDC+TaskDecorator

自定义MDCTaskDecorator

/**
 * 线程池修饰类
 */
public class MDCTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        // 获取主线程中的请求信息(我们的用户信息也放在里面)
        String userId = MDC.get("userId");
        Map<String, String> copyOfContextMap = MDC.getCopyOfContextMap();
        System.out.println(copyOfContextMap);
        return () -> {
            try {
                // 将主线程的请求信息,设置到子线程中
                MDC.put("userId",userId);
                // 执行子线程,这一步不要忘了
                runnable.run();
            } finally {
                // 线程结束,清空这些信息,否则可能造成内存泄漏
                MDC.clear();
            }
        };
    }
}
复制代码
ExecutorConfig

在原来的基础上增加 executor.setTaskDecorator(new MDCTaskDecorator());

@Bean(name = "asyncServiceExecutor")
    public Executor asyncServiceExecutor() {
        log.info("start asyncServiceExecutor----------------");
        //ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //使用可视化运行状态的线程池
        ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(corePoolSize);
        //配置最大线程数
        executor.setMaxPoolSize(maxPoolSize);
        //配置队列大小
        executor.setQueueCapacity(queueCapacity);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix(namePrefix);

        // rejection-policy:当pool已经达到max size的时候,如何处理新任务
        // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        //增加MDC的线程池修饰类
        executor.setTaskDecorator(new MDCTaskDecorator());
        //执行初始化
        executor.initialize();
        log.info("end asyncServiceExecutor------------");
        return executor;
    }
复制代码
AsyncServiceImpl
/**
     * 使用MDC获取主线程传递的数据
     * @return
     * @throws InterruptedException
     */
    @Async("asyncServiceExecutor")
    public CompletableFuture<String> executeValueAsync5() throws InterruptedException {
        log.info("start executeValueAsync");
        System.out.println("异步线程执行返回结果......+");
        log.info("end executeValueAsync");
        return CompletableFuture.completedFuture(MDC.get("userId"));
    }


复制代码
Test2Controller
/**
     * 使用MDC+TaskDecorator方式
     * 本质也是ThreadLocal+TaskDecorator方式
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @GetMapping("/test5")
    public String test5() throws InterruptedException, ExecutionException {
        MDC.put("userId","123456");
        CompletableFuture<String> completableFuture = asyncService.executeValueAsync5();
        String s = completableFuture.get();
        return s;
    }
复制代码

姿势4:InheritableThreadLocal

用户工具类 UserInheritableUtils

//**
 *使用InheritableThreadLocal存储线程之间共享的数据变量,如登录的用户信息
 */
public class UserInheritableUtils {
    private static  final  InheritableThreadLocal<String> userLocal=new InheritableThreadLocal<>();

    public static  String getUserId(){
        return userLocal.get();
    }
    public static void setUserId(String userId){
        userLocal.set(userId);
    }

    public static void clear(){
        userLocal.remove();
    }

}
复制代码
AsyncServiceImpl
/**
     * 使用InheritableThreadLocal获取主线程传递的数据
     * @return
     * @throws InterruptedException
     */
    @Async("asyncServiceExecutor")
    public CompletableFuture<String> executeValueAsync4() throws InterruptedException {
        log.info("start executeValueAsync");
        System.out.println("异步线程执行返回结果......+");
        log.info("end executeValueAsync");
        return CompletableFuture.completedFuture(UserInheritableUtils.getUserId());
    }

复制代码
Test2Controller
/**
     * 使用InheritableThreadLocal方式
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @GetMapping("/test4")
    public String test4(@RequestParam("userId") String userId) throws InterruptedException, ExecutionException {
        UserInheritableUtils.setUserId(userId);
        CompletableFuture<String> completableFuture = asyncService.executeValueAsync4();
        String s = completableFuture.get();
        return s;
    }
复制代码

姿势5:TransmittableThreadLocal

用户工具类 UserTransmittableUtils

/**
 *使用TransmittableThreadLocal存储线程之间共享的数据变量,如登录的用户信息
 */
public class UserTransmittableUtils {
    private static  final TransmittableThreadLocal<String> userLocal=new TransmittableThreadLocal<>();

    public static  String getUserId(){
        return userLocal.get();
    }
    public static void setUserId(String userId){
        userLocal.set(userId);
    }

    public static void clear(){
        userLocal.remove();
    }

}

}
复制代码
AsyncServiceImpl
/**
     * 使用TransmittableThreadLocal获取主线程传递的数据
     * @return
     * @throws InterruptedException
     */
    @Async("asyncServiceExecutor")
    public CompletableFuture<String> executeValueAsync6() throws InterruptedException {
        log.info("start executeValueAsync");
        System.out.println("异步线程执行返回结果......+");
        log.info("end executeValueAsync");
        return CompletableFuture.completedFuture(UserTransmittableUtils.getUserId());
    }

复制代码
Test2Controller
/**
     * 使用TransmittableThreadLocal方式
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @GetMapping("/test6")
    public String test6() throws InterruptedException, ExecutionException {
        UserTransmittableUtils.setUserId("123456");
        CompletableFuture<String> completableFuture = asyncService.executeValueAsync6();
        String s = completableFuture.get();
        return s;
    }
复制代码

maven依赖

<dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>transmittable-thread-local</artifactId>
            <version>2.12.1</version>
        </dependency>
复制代码

方案对比

方案1,方案2,方案3主要是借助TaskDecorator进行父子线程之间传递数据。其中MDC方案主要借鉴于MDC的日志跟踪的思想来实现,关于MDC相关的日志跟踪后续会学习分享

方案4和方案5使用InheritableThreadLocal和TransmittableThreadLocal来实现,其中TransmittableThreadLocal是阿里InheritableThreadLocal进行优化封装。为什么要封装,有兴趣的可以去学习《 加强版ThreadLocal之阿里开源TransmittableThreadLocal学习 》

本人推荐使用方案5,哈哈。

简答说一下InheritableThreadLocal

public static void mAIn(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1,1,1,
                TimeUnit.MINUTES,new ArrayBlockingQueue<>(1));

        ThreadLocal local = new InheritableThreadLocal();
        local.set(1);

        executor.execute(()->{
            System.out.println("打印1:"+local.get());
        });
        local.set(2);

        System.out.println("打印2:"+local.get());

        executor.execute(()->{
            System.out.println("打印3:"+local.get());
        });
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("打印4:"+local.get());
            }
        }).start();
    }
复制代码

运行结果如下

打印2:2
打印1:1
打印3:1
打印4:2
复制代码

分析: 分析打印3为什么是1,InheritableThreadLocal的继承性是在new Thread创建子线程时候在构造函数内把父线程内线程变量拷贝到子线程内部的。为了不在创建新线程耗费资源,我们一般会用线程池,线程池的线程会复用,那么线程中的ThreadLocal便不对了,可能是旧的,因为线程是旧的。



Tags:SpringBoot   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,不构成投资建议。投资者据此操作,风险自担。如有任何标注错误或版权侵犯请与我们联系,我们将及时更正、删除。
▌相关推荐
详解基于SpringBoot的WebSocket应用开发
在现代Web应用中,实时交互和数据推送的需求日益增长。WebSocket协议作为一种全双工通信协议,允许服务端与客户端之间建立持久性的连接,实现实时、双向的数据传输,极大地提升了用...【详细内容】
2024-01-30  Search: SpringBoot  点击:(19)  评论:(0)  加入收藏
SpringBoot如何实现缓存预热?
缓存预热是指在 Spring Boot 项目启动时,预先将数据加载到缓存系统(如 Redis)中的一种机制。那么问题来了,在 Spring Boot 项目启动之后,在什么时候?在哪里可以将数据加载到缓存系...【详细内容】
2024-01-19  Search: SpringBoot  点击:(86)  评论:(0)  加入收藏
SpringBoot3+Vue3 开发高并发秒杀抢购系统
开发高并发秒杀抢购系统:使用SpringBoot3+Vue3的实践之旅随着互联网技术的发展,电商行业对秒杀抢购系统的需求越来越高。为了满足这种高并发、高流量的场景,我们决定使用Spring...【详细内容】
2024-01-14  Search: SpringBoot  点击:(91)  评论:(0)  加入收藏
公司用了六年的 SpringBoot 项目部署方案,稳得一批!
本篇和大家分享的是springboot打包并结合shell脚本命令部署,重点在分享一个shell程序启动工具,希望能便利工作。 profiles指定不同环境的配置 maven-assembly-plugin打发布压...【详细内容】
2024-01-10  Search: SpringBoot  点击:(165)  评论:(0)  加入收藏
简易版的SpringBoot是如何实现的!!!
SpringBoot作为目前最流行的框架之一,同时是每个程序员必须掌握的知识,其提供了丰富的功能模块和开箱即用的特性,极大地提高了开发效率和降低了学习成本,使得开发人员能够更专注...【详细内容】
2023-12-29  Search: SpringBoot  点击:(136)  评论:(0)  加入收藏
用 SpringBoot+Redis 解决海量重复提交问题
前言 一:搭建redis的服务Api 二:自定义注解AutoIdempotent 三:token创建和检验 四:拦截器的配置 五:测试用例 六:总结前言:在实际的开发项目中,一个对外暴露的接口往往会面临很多...【详细内容】
2023-12-20  Search: SpringBoot  点击:(53)  评论:(0)  加入收藏
SpringBoot中如何优雅地个性化定制Jackson
当使用 JSON 格式时,Spring Boot 将使用ObjectMapper实例来序列化响应和反序列化请求。在本教程中,我们将了解配置序列化和反序列化选项的最常用方法。一、默认配置默认情况下...【详细内容】
2023-12-20  Search: SpringBoot  点击:(132)  评论:(0)  加入收藏
springboot-如何集成Validation进行参数校验
一、步骤概览 二、步骤说明1.引入依赖包在 pom.xml 文件中引入 validation 组件,它提供了在 Spring Boot 应用程序中进行参数校验的支持。<!-- WEB 程序依赖包 --><dependen...【详细内容】
2023-12-13  Search: SpringBoot  点击:(157)  评论:(0)  加入收藏
优雅的springboot参数校验,你学会了吗?
前言在后端的接口开发过程,实际上每一个接口都或多或少有不同规则的参数校验,有一些是基础校验,如非空校验、长度校验、大小校验、格式校验;也有一些校验是业务校验,如学号不能重...【详细内容】
2023-11-29  Search: SpringBoot  点击:(200)  评论:(0)  加入收藏
Springboot扩展点之BeanDefinitionRegistryPostProcessor,你学会了吗?
前言通过这篇文章来大家分享一下,另外一个Springboot的扩展点BeanDefinitionRegistryPostProcessor,一般称这类扩展点为容器级后置处理器,另外一类是Bean级的后置处理器;容器级...【详细内容】
2023-11-27  Search: SpringBoot  点击:(175)  评论:(0)  加入收藏
▌简易百科推荐
Qt与Flutter:在跨平台UI框架中哪个更受欢迎?
在跨平台UI框架领域,Qt和Flutter是两个备受瞩目的选择。它们各自具有独特的优势,也各自有着广泛的应用场景。本文将对Qt和Flutter进行详细的比较,以探讨在跨平台UI框架中哪个更...【详细内容】
2024-04-12  刘长伟    Tags:UI框架   点击:(1)  评论:(0)  加入收藏
Web Components实践:如何搭建一个框架无关的AI组件库
一、让人又爱又恨的Web ComponentsWeb Components是一种用于构建可重用的Web元素的技术。它允许开发者创建自定义的HTML元素,这些元素可以在不同的Web应用程序中重复使用,并且...【详细内容】
2024-04-03  京东云开发者    Tags:Web Components   点击:(8)  评论:(0)  加入收藏
Kubernetes 集群 CPU 使用率只有 13% :这下大家该知道如何省钱了
作者 | THE STACK译者 | 刘雅梦策划 | Tina根据 CAST AI 对 4000 个 Kubernetes 集群的分析,Kubernetes 集群通常只使用 13% 的 CPU 和平均 20% 的内存,这表明存在严重的过度...【详细内容】
2024-03-08  InfoQ    Tags:Kubernetes   点击:(17)  评论:(0)  加入收藏
Spring Security:保障应用安全的利器
SpringSecurity作为一个功能强大的安全框架,为Java应用程序提供了全面的安全保障,包括认证、授权、防护和集成等方面。本文将介绍SpringSecurity在这些方面的特性和优势,以及它...【详细内容】
2024-02-27  风舞凋零叶    Tags:Spring Security   点击:(54)  评论:(0)  加入收藏
五大跨平台桌面应用开发框架:Electron、Tauri、Flutter等
一、什么是跨平台桌面应用开发框架跨平台桌面应用开发框架是一种工具或框架,它允许开发者使用一种统一的代码库或语言来创建能够在多个操作系统上运行的桌面应用程序。传统上...【详细内容】
2024-02-26  贝格前端工场    Tags:框架   点击:(47)  评论:(0)  加入收藏
Spring Security权限控制框架使用指南
在常用的后台管理系统中,通常都会有访问权限控制的需求,用于限制不同人员对于接口的访问能力,如果用户不具备指定的权限,则不能访问某些接口。本文将用 waynboot-mall 项目举例...【详细内容】
2024-02-19  程序员wayn  微信公众号  Tags:Spring   点击:(39)  评论:(0)  加入收藏
开发者的Kubernetes懒人指南
你可以将本文作为开发者快速了解 Kubernetes 的指南。从基础知识到更高级的主题,如 Helm Chart,以及所有这些如何影响你作为开发者。译自Kubernetes for Lazy Developers。作...【详细内容】
2024-02-01  云云众生s  微信公众号  Tags:Kubernetes   点击:(51)  评论:(0)  加入收藏
链世界:一种简单而有效的人类行为Agent模型强化学习框架
强化学习是一种机器学习的方法,它通过让智能体(Agent)与环境交互,从而学习如何选择最优的行动来最大化累积的奖励。强化学习在许多领域都有广泛的应用,例如游戏、机器人、自动驾...【详细内容】
2024-01-30  大噬元兽  微信公众号  Tags:框架   点击:(68)  评论:(0)  加入收藏
Spring实现Kafka重试Topic,真的太香了
概述Kafka的强大功能之一是每个分区都有一个Consumer的偏移值。该偏移值是消费者将读取的下一条消息的值。可以自动或手动增加该值。如果我们由于错误而无法处理消息并想重...【详细内容】
2024-01-26  HELLO程序员  微信公众号  Tags:Spring   点击:(88)  评论:(0)  加入收藏
SpringBoot如何实现缓存预热?
缓存预热是指在 Spring Boot 项目启动时,预先将数据加载到缓存系统(如 Redis)中的一种机制。那么问题来了,在 Spring Boot 项目启动之后,在什么时候?在哪里可以将数据加载到缓存系...【详细内容】
2024-01-19   Java中文社群  微信公众号  Tags:SpringBoot   点击:(86)  评论:(0)  加入收藏
站内最新
站内热门
站内头条