文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

SpringBoot异步线程间数据传递的四种方式

2023-01-06 15:00

关注

前面在介绍分布式链路追踪时讲过异步调用会丢失链路信息，最终的解决方案是使用对应的包装类重新包装一下，如下：

还有openFeign异步请求丢失上文的问题，这些问题追根究底都是ThreadLocal惹得祸。

由于ThreadLocal只能保存当前线程的信息，不能实现父子线程的继承。

说到这，很多人想到了InheritableThreadLocal，确实InheritableThreadLocal能够实现父子线程间传递本地变量，但是.....

但是你的程序如果采用线程池，则存在着线程复用的情况，这时就不一定能够实现父子线程间传递了，因为在线程在线程池中的存在不是每次使用都会进行创建，InheritableThreadlocal是在线程初始化时intertableThreadLocals=true才会进行拷贝传递。

所以若本次使用的子线程是已经被池化的线程，从线程池中取出线下进行使用，是没有经过初始化的过程，也就不会进行父子线程的本地变量拷贝。

由于在日常应用场景中，绝大多数都是会采用线程池的方式进行资源的有效管理。

今天就来聊一聊阿里的ThansmittableThreadLocal是如何解决线程池中父子线程本地变量传递。

B站链接：https://b23.tv/RI06iZl

下方是本篇文章视频教程，讲解更加详细！图片

进度条，百分之9

“

帅气的人都点赞了~

”

InheritableThreadLocal 的问题

在介绍ThansmittableThreadLocal之前先来看一下InheritableThreadLocal 在线程池中的问题，如下代码：

@Test
public void test() throws Exception {
    //单一线程池
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    //InheritableThreadLocal存储
    InheritableThreadLocal<String> username = new InheritableThreadLocal<>();
    for (int i = 0; i < 10; i++) {
    username.set("公众号：码猿技术专栏—"+i);
    Thread.sleep(3000);
    CompletableFuture.runAsync(()-> System.out.println(username.get()),executorService);
   }
}

上述代码中创建了一个单一线程池，循环异步调用，打印一下username，由于核心线程数是1，势必存在线程的复用。

打印信息如下：

公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0
公众号：码猿技术专栏—0

看到了吗？这里并没有实现父子线程间的变量传递，这也就是InheritableThreadLocal 的局限性。

TransmittableThreadLocal 使用

TransmittableThreadLocal(TTL)：在使用线程池等会池化复用线程的执行组件情况下，提供ThreadLocal值的传递功能，解决异步执行时上下文传递的问题。

整个TransmittableThreadLocal库的核心功能（用户API与框架/中间件的集成API、线程池ExecutorService/ForkJoinPool/TimerTask及其线程工厂的Wrapper）。

需求场景：

“

官网地址：https://github.com/alibaba/transmittable-thread-local

”

下面就以上面的例子改造成TransmittableThreadLocal试一下效果。

首选需要引入对应的依赖，如下：

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>transmittable-thread-local</artifactId>
</dependency

改造后的代码如下：

@Test
public void test() throws Exception {
    //单一线程池
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    //需要使用TtlExecutors对线程池包装一下
    executorService=TtlExecutors.getTtlExecutorService(executorService);
    //TransmittableThreadLocal创建
    TransmittableThreadLocal<String> username = new TransmittableThreadLocal<>();
    for (int i = 0; i < 10; i++) {
    username.set("公众号：码猿技术专栏—"+i);
    Thread.sleep(3000);
    CompletableFuture.runAsync(()-> System.out.println(username.get()),executorService);
  }
}

需要注意的是需要使用TtlExecutors对线程池进行包装，代码如下：

executorService=TtlExecutors.getTtlExecutorService(executorService);

运行效果如下：

公众号：码猿技术专栏—0
公众号：码猿技术专栏—1
公众号：码猿技术专栏—2
公众号：码猿技术专栏—3
公众号：码猿技术专栏—4
公众号：码猿技术专栏—5
公众号：码猿技术专栏—6
公众号：码猿技术专栏—7
公众号：码猿技术专栏—8
公众号：码猿技术专栏—9

可以看到已经能够实现了线程池中的父子线程的数据传递。

在每次调用任务的时，都会将当前的主线程的TTL数据copy到子线程里面，执行完成后，再清除掉。同时子线程里面的修改回到主线程时其实并没有生效。这样可以保证每次任务执行的时候都是互不干涉。

简单应用

在 Spring Security 往往需要存储用户登录的详细信息，这样在业务方法中能够随时获取用户的信息。

在前面的Spring Cloud Gateway整合OAuth2.0实现统一认证鉴权 文章中笔者是将用户信息直接存储在Request中，这样每次请求都能获取到对应的信息。

其实Request中的信息存储也是通过ThreadLocal完成的，在异步执行的时候还是需要重新转存，这样一来代码就变得复杂。

那么了解了TransmittableThreadLocal 之后，完全可以使用这个存储用户的登录信息，实现如下：


public class SecurityContextHolder {

    //使用TTL存储身份信息
    private static final TransmittableThreadLocal<LoginVal> THREAD_LOCAL = new TransmittableThreadLocal<>();

    public static void set(LoginVal loginVal){
        THREAD_LOCAL.set(loginVal);
    }

    public static LoginVal get(){
        return THREAD_LOCAL.get();
    }

    public static void remove(){
        THREAD_LOCAL.remove();
    }

}

由于mvc中的一次请求对应一个线程，因此只需要在拦截器中的设置和移除TransmittableThreadLocal中的信息，代码如下：


@Component
public class AuthInterceptor implements AsyncHandlerInterceptor {
    
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        if (!(handler instanceof HandlerMethod))
            return true;
        //获取请求头中的加密的用户信息
        String token = request.getHeader(OAuthConstant.TOKEN_NAME);
        if (StrUtil.isBlank(token))
            return true;
        //解密
        String json = Base64.decodeStr(token);
        //将json解析成LoginVal
        LoginVal loginVal = TokenUtils.parseJsonToLoginVal(json);
        //封装数据到ThreadLocal中
        SecurityContextHolder.set(loginVal);
        return true;
    }

    
    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex){
        SecurityContextHolder.remove();
    }
}

原理

从定义来看，TransimittableThreadLocal继承于InheritableThreadLocal，并实现TtlCopier接口，它里面只有一个copy方法。所以主要是对InheritableThreadLocal的扩展。

public class TransmittableThreadLocal<T> extends InheritableThreadLocal<T> implements TtlCopier<T> 

在TransimittableThreadLocal中添加holder属性。这个属性的作用就是被标记为具备线程传递资格的对象都会被添加到这个对象中。

要标记一个类，比较容易想到的方式，就是给这个类新增一个Type字段，还有一个方法就是将具备这种类型的的对象都添加到一个静态全局集合中。之后使用时，这个集合里的所有值都具备这个标记。

// 1. holder本身是一个InheritableThreadLocal对象
// 2. 这个holder对象的value是WeakHashMap<TransmittableThreadLocal<Object>, ?>
//   2.1 WeekHashMap的value总是null,且不可能被使用。
//    2.2 WeekHasshMap支持value=null
private static InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>> holder = new InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>>() {
  @Override
  protected WeakHashMap<TransmittableThreadLocal<Object>, ?> initialValue() {
    return new WeakHashMap<TransmittableThreadLocal<Object>, Object>();
  }
 
  
  @Override
  protected WeakHashMap<TransmittableThreadLocal<Object>, ?> childValue(WeakHashMap<TransmittableThreadLocal<Object>, ?> parentValue) {
    return new WeakHashMap<TransmittableThreadLocal<Object>, Object>(parentValue);
  }
};

应用代码是通过TtlExecutors工具类对线程池对象进行包装。工具类只是简单的判断，输入的线程池是否已经被包装过、非空校验等，然后返回包装类ExecutorServiceTtlWrapper。根据不同的线程池类型，有不同和的包装类。

@Nullable
public static ExecutorService getTtlExecutorService(@Nullable ExecutorService executorService) {
  if (TtlAgent.isTtlAgentLoaded() || executorService == null || executorService instanceof TtlEnhanced) {
    return executorService;
  }
  return new ExecutorServiceTtlWrapper(executorService);
}

进入包装类ExecutorServiceTtlWrapper。可以注意到不论是通过ExecutorServiceTtlWrapper#submit方法或者是ExecutorTtlWrapper#execute方法，都会将线程对象包装成TtlCallable或者TtlRunnable，用于在真正执行run方法前做一些业务逻辑。


@NonNull
@Override
public <T> Future<T> submit(@NonNull Callable<T> task) {
  return executorService.submit(TtlCallable.get(task));
}


@Override
public void execute(@NonNull Runnable command) {
  executor.execute(TtlRunnable.get(command));
}

所以，重点的核心逻辑应该是在TtlCallable#call()或者TtlRunnable#run()中。以下以TtlCallable为例，TtlRunnable同理类似。在分析call()方法之前，先看一个类Transmitter

public static class Transmitter {
  
  @NonNull
  public static Object capture() {
    return new Snapshot(captureTtlValues(), captureThreadLocalValues());
  }
 
    
  private static HashMap<TransmittableThreadLocal<Object>, Object> captureTtlValues() {
    HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value = 
      new HashMap<TransmittableThreadLocal<Object>, Object>();
    for (TransmittableThreadLocal<Object> threadLocal : holder.get().keySet()) {
      ttl2Value.put(threadLocal, threadLocal.copyValue());
    }
    return ttl2Value;
  }

  
  private static HashMap<ThreadLocal<Object>, Object> captureThreadLocalValues() {
    final HashMap<ThreadLocal<Object>, Object> threadLocal2Value = 
      new HashMap<ThreadLocal<Object>, Object>();
    for(Map.Entry<ThreadLocal<Object>,TtlCopier<Object>>entry:threadLocalHolder.entrySet()){
      final ThreadLocal<Object> threadLocal = entry.getKey();
      final TtlCopier<Object> copier = entry.getValue();
      threadLocal2Value.put(threadLocal, copier.copy(threadLocal.get()));
    }
    return threadLocal2Value;
  }

  
  @NonNull
  public static Object replay(@NonNull Object captured) {
    final Snapshot capturedSnapshot = (Snapshot) captured;
    return new Snapshot(replayTtlValues(capturedSnapshot.ttl2Value), 
                        replayThreadLocalValues(capturedSnapshot.threadLocal2Value));
  }
 
  
  @NonNull
  private static HashMap<TransmittableThreadLocal<Object>, Object> replayTtlValues(@NonNull HashMap<TransmittableThreadLocal<Object>, Object> captured) {
    // 创建副本backup
    HashMap<TransmittableThreadLocal<Object>, Object> backup = 
      new HashMap<TransmittableThreadLocal<Object>, Object>();

    for (final Iterator<TransmittableThreadLocal<Object>> iterator = holder.get().keySet().iterator(); iterator.hasNext(); ) {
      TransmittableThreadLocal<Object> threadLocal = iterator.next();
      // 对当前线程的本地变量进行副本拷贝
      backup.put(threadLocal, threadLocal.get());

      // 若出现调用线程中不存在某个线程变量，而线程池中线程有，则删除线程池中对应的本地变量
      if (!captured.containsKey(threadLocal)) {
        iterator.remove();
        threadLocal.superRemove();
      }
    }
    // 将捕获的TTL值打入线程池获取到的线程TTL中。
    setTtlValuesTo(captured);
    // 是一个扩展点，调用TTL的beforeExecute方法。默认实现为空
    doExecuteCallback(true);
    return backup;
  }

  private static HashMap<ThreadLocal<Object>, Object> replayThreadLocalValues(@NonNull HashMap<ThreadLocal<Object>, Object> captured) {
    final HashMap<ThreadLocal<Object>, Object> backup = 
      new HashMap<ThreadLocal<Object>, Object>();
    for (Map.Entry<ThreadLocal<Object>, Object> entry : captured.entrySet()) {
      final ThreadLocal<Object> threadLocal = entry.getKey();
      backup.put(threadLocal, threadLocal.get());
      final Object value = entry.getValue();
      if (value == threadLocalClearMark) threadLocal.remove();
      else threadLocal.set(value);
    }
    return backup;
  }

  
  @NonNull
  public static Object clear() {
    final HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value = 
      new HashMap<TransmittableThreadLocal<Object>, Object>();

    final HashMap<ThreadLocal<Object>, Object> threadLocal2Value = 
      new HashMap<ThreadLocal<Object>, Object>();
    for(Map.Entry<ThreadLocal<Object>,TtlCopier<Object>>entry:threadLocalHolder.entrySet()){
      final ThreadLocal<Object> threadLocal = entry.getKey();
      threadLocal2Value.put(threadLocal, threadLocalClearMark);
    }
    return replay(new Snapshot(ttl2Value, threadLocal2Value));
  }

  
  public static void restore(@NonNull Object backup) {
    final Snapshot backupSnapshot = (Snapshot) backup;
    restoreTtlValues(backupSnapshot.ttl2Value);
    restoreThreadLocalValues(backupSnapshot.threadLocal2Value);
  }

  private static void restoreTtlValues(@NonNull HashMap<TransmittableThreadLocal<Object>, Object> backup) {
    // 扩展点，调用TTL的afterExecute
    doExecuteCallback(false);

    for (final Iterator<TransmittableThreadLocal<Object>> iterator = holder.get().keySet().iterator(); iterator.hasNext(); ) {
      TransmittableThreadLocal<Object> threadLocal = iterator.next();

      if (!backup.containsKey(threadLocal)) {
        iterator.remove();
        threadLocal.superRemove();
      }
    }

    // 将本地变量恢复成备份版本
    setTtlValuesTo(backup);
  }

  private static void setTtlValuesTo(@NonNull HashMap<TransmittableThreadLocal<Object>, Object> ttlValues) {
    for (Map.Entry<TransmittableThreadLocal<Object>, Object> entry : ttlValues.entrySet()) {
      TransmittableThreadLocal<Object> threadLocal = entry.getKey();
      threadLocal.set(entry.getValue());
    }
  }

  private static void restoreThreadLocalValues(@NonNull HashMap<ThreadLocal<Object>, Object> backup) {
    for (Map.Entry<ThreadLocal<Object>, Object> entry : backup.entrySet()) {
      final ThreadLocal<Object> threadLocal = entry.getKey();
      threadLocal.set(entry.getValue());
    }
  }

  
  private static class Snapshot {
    final HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value;
    final HashMap<ThreadLocal<Object>, Object> threadLocal2Value;

    private Snapshot(HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value,
                     HashMap<ThreadLocal<Object>, Object> threadLocal2Value) {
      this.ttl2Value = ttl2Value;
      this.threadLocal2Value = threadLocal2Value;
    }
  }

进入TtlCallable#call()方法。

@Override
public V call() throws Exception {
  Object captured = capturedRef.get();
  if (captured == null || releaseTtlValueReferenceAfterCall &amp;&amp; 
      !capturedRef.compareAndSet(captured, null)) {
    throw new IllegalStateException("TTL value reference is released after call!");
  }
  // 调用replay方法将捕获到的当前线程的本地变量，传递给线程池线程的本地变量，
  // 并且获取到线程池线程覆盖之前的本地变量副本。
  Object backup = replay(captured);
  try {
    // 线程方法调用
    return callable.call();
  } finally {
    // 使用副本进行恢复。
    restore(backup);
  }
}

到这基本上线程池方式传递本地变量的核心代码已经大概看完了。总的来说在创建TtlCallable对象是，调用capture()方法捕获调用方的本地线程变量，在call()执行时，将捕获到的线程变量，替换到线程池所对应获取到的线程的本地变量中，并且在执行完成之后，将其本地变量恢复到调用之前。

总结

本文介绍了使用阿里开源的TransmittableThreadLocal 优雅的实现父子线程的数据传递，应用场景很多，企业中应用也比较广泛。

阅读原文内容投诉

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）

    难度     813人已做
    查看

  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看

  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看

  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）

    难度     435人已做
    查看

  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机

SpringBoot异步线程间数据传递的四种方式

后端开发2023-01-06

Spring Boot异步线程间数据传递的四种方式

后端开发2023-01-06

SpringBoot 异步线程间传递上下文方式

后端开发2024-04-02

SpringBoot 异步线程间传递上下文方式是怎样的

后端开发2023-06-25

用这四招优雅地实现 Spring Boot 异步线程间数据传递

后端开发Spring异步2024-12-13

SpringBoot父子线程数据传递的五种方案介绍

后端开发2024-04-02

这样实现异步线程间 数据传递，太优雅了！

后端开发异步线程间2024-11-28

ASP.NETMVC视图页使用jQuery传递异步数据的几种方式详解

后端开发2024-04-02

关于Android Activity之间传递数据的6种方式

后端开发2022-06-06

linux线程间传递数据的方法是什么

后端开发2023-09-26

PHP 函数的参数传递方式在异步编程中的运用？

PHP 函数的参数传递方式在异步编程中的运用？
后端开发2024-04-15

小程序页面之间数据传递的4种方法总结

后端开发2023-05-17
位置：首页-资讯-后端开发
咦！没有更多了？去看看其它编程学习网 内容吧
首页课程
资料下载问答资讯