关于springboot:用这4招优雅的实现Spring-Boot-异步线程间数据传递

Spring Boot 自定义线程池实现异步开发置信大家都理解，然而在理论开发中须要在父子线程之间传递一些数据，比方用户信息，链路信息等等

比方用户登录信息应用ThreadLocal寄存保障线程隔离，代码如下：

/**
 * @author 公众号：码猿技术专栏
 * @description 用户上下文信息
 */
public class OauthContext {
    private static  final  ThreadLocal<LoginVal> loginValThreadLocal=new ThreadLocal<>();

    public static  LoginVal get(){
        return loginValThreadLocal.get();
    }
    public static void set(LoginVal loginVal){
        loginValThreadLocal.set(loginVal);
    }
    public static void clear(){
        loginValThreadLocal.remove();
    }
}

那么子线程想要获取这个LoginVal如何做呢？

明天就来介绍几种优雅的形式实现Spring Boot 外部的父子线程的数据传递。

1. 手动设置

每执行一次异步线程都要分为两步：

获取父线程的LoginVal
将LoginVal设置到子线程，达到复用

代码如下：

public void handlerAsync() {
        //1\. 获取父线程的loginVal
        LoginVal loginVal = OauthContext.get();
        log.info("父线程的值：{}",OauthContext.get());
        CompletableFuture.runAsync(()->{
            //2\. 设置子线程的值，复用
           OauthContext.set(loginVal);
           log.info("子线程的值：{}",OauthContext.get());
        });
    }

尽管可能实现目标，然而每次开异步线程都须要手动设置，反复代码太多，看了头疼，你认为优雅吗？

2. 线程池设置TaskDecorator

TaskDecorator是什么？官网api的大抵意思：这是一个执行回调办法的装璜器，次要利用于传递上下文，或者提供工作的监控/统计信息。

晓得有这么一个货色，如何去应用？

TaskDecorator是一个接口，首先须要去实现它，代码如下：

/**
 * @author 公众号：码猿技术专栏
 * @description 上下文装璜器
 */
public class ContextTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        //获取父线程的loginVal
        LoginVal loginVal = OauthContext.get();
        return () -> {
            try {
                // 将主线程的申请信息，设置到子线程中
                OauthContext.set(loginVal);
                // 执行子线程，这一步不要忘了
                runnable.run();
            } finally {
                // 线程完结，清空这些信息，否则可能造成内存透露
                OauthContext.clear();
            }
        };
    }
}

这里我只是设置了LoginVal，理论开发中其余的共享数据，比方SecurityContext，RequestAttributes….

TaskDecorator须要联合线程池应用，理论开发中异步线程倡议应用线程池，只须要在对应的线程池配置一下，代码如下：

@Bean("taskExecutor")
public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
        poolTaskExecutor.setCorePoolSize(xx);
        poolTaskExecutor.setMaxPoolSize(xx);
        // 设置线程沉闷工夫（秒）
        poolTaskExecutor.setKeepAliveSeconds(xx);
        // 设置队列容量
        poolTaskExecutor.setQueueCapacity(xx);
        //设置TaskDecorator，用于解决父子线程间的数据复用
        poolTaskExecutor.setTaskDecorator(new ContextTaskDecorator());
        poolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 期待所有工作完结后再敞开线程池
        poolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        return poolTaskExecutor;
    }

此时业务代码就不须要去设置子线程的值，间接应用即可，代码如下：

public void handlerAsync() {
        log.info("父线程的用户信息：{}", OauthContext.get());
        //执行异步工作，须要指定的线程池
        CompletableFuture.runAsync(()-> log.info("子线程的用户信息：{}", OauthContext.get()),taskExecutor);
    }

来看一下后果，如下图：

这里应用的是CompletableFuture执行异步工作，应用@Async这个注解同样是可行的。

留神：无论应用何种形式，都须要指定线程池

3. InheritableThreadLocal

这种计划不倡议应用，InheritableThreadLocal尽管可能实现父子线程间的复用，然而在线程池中应用会存在复用的问题，具体的能够看陈某之前的文章：微服务中应用阿里开源的TTL，优雅的实现身份信息的线程间复用

这种计划应用也是非常简单，间接用InheritableThreadLocal替换ThreadLocal即可，代码如下：

/**
 * @author 公众号：码猿技术专栏
 * @description 用户上下文信息
 */
public class OauthContext {
    private static  final  InheritableThreadLocal<LoginVal> loginValThreadLocal=new InheritableThreadLocal<>();

    public static  LoginVal get(){
        return loginValThreadLocal.get();
    }
    public static void set(LoginVal loginVal){
        loginValThreadLocal.set(loginVal);
    }
    public static void clear(){
        loginValThreadLocal.remove();
    }
}

4. TransmittableThreadLocal

TransmittableThreadLocal是阿里开源的工具，补救了InheritableThreadLocal的缺点，在应用线程池等会池化复用线程的执行组件状况下，提供ThreadLocal值的传递性能，解决异步执行时上下文传递的问题。

应用起来也是非常简单，增加依赖如下：

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

OauthContext革新代码如下：

/**
 * @author 公众号：码猿技术专栏
 * @description 用户上下文信息
 */
public class OauthContext {
    private static  final TransmittableThreadLocal<LoginVal> loginValThreadLocal=new TransmittableThreadLocal<>();

    public static  LoginVal get(){
        return loginValThreadLocal.get();
    }
    public static void set(LoginVal loginVal){
        loginValThreadLocal.set(loginVal);
    }
    public static void clear(){
        loginValThreadLocal.remove();
    }
}

对于TransmittableThreadLocal想深刻理解其原理能够看陈某之前的文章：微服务中应用阿里开源的TTL，优雅的实现身份信息的线程间复用，利用还是十分宽泛的

TransmittableThreadLocal原理

从定义来看，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>();
  }

  /**
   * 重写了childValue办法，实现上间接将父线程的属性作为子线程的本地变量对象。
   */
  @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办法前做一些业务逻辑。

/**
 * 在ExecutorServiceTtlWrapper实现submit办法
 */
@NonNull
@Override
public <T> Future<T> submit(@NonNull Callable<T> task) {
  return executorService.submit(TtlCallable.get(task));
}

/**
 * 在ExecutorTtlWrapper实现execute办法
 */
@Override
public void execute(@NonNull Runnable command) {
  executor.execute(TtlRunnable.get(command));
}

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

public static class Transmitter {
  /**
    * 捕捉以后线程中的是所有TransimittableThreadLocal和注册ThreadLocal的值。
    */
  @NonNull
  public static Object capture() {
    return new Snapshot(captureTtlValues(), captureThreadLocalValues());
  }

    /**
    * 捕捉TransimittableThreadLocal的值,将holder中的所有值都增加到HashMap后返回。
    */
  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;
  }

  /**
    * 捕捉注册的ThreadLocal的值,也就是本来线程中的ThreadLocal,能够注册到TTL中，在
    * 进行线程池本地变量传递时也会被传递。
    */
  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;
  }

  /**
    * 将捕捉到的本地变量进行替换子线程的本地变量，并且返回子线程现有的本地变量正本backup。
    * 用于在执行run/call办法之后，将本地变量正本复原。
    */
  @NonNull
  public static Object replay(@NonNull Object captured) {
    final Snapshot capturedSnapshot = (Snapshot) captured;
    return new Snapshot(replayTtlValues(capturedSnapshot.ttl2Value), 
                        replayThreadLocalValues(capturedSnapshot.threadLocal2Value));
  }

  /**
    * 替换TransmittableThreadLocal
    */
  @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;
  }

  /**
    * 革除复线线程的所有TTL和TL，并返回革除之气的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());
    }
  }

  /**
   * 快照类，保留TTL和TL
   */
  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 && 
      !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()执行时，将捕捉到的线程变量，替换到线程池所对应获取到的线程的本地变量中，并且在执行实现之后，将其本地变量复原到调用之前。

总结

上述列举了4种计划，这里举荐计划2和计划4，其中两种计划的毛病非常明显，理论开发中也是采纳的计划2或者计划4

关于springboot:用这4招优雅的实现Spring-Boot-异步线程间数据传递

1. 手动设置

2. 线程池设置TaskDecorator

3. InheritableThreadLocal

4. TransmittableThreadLocal

TransmittableThreadLocal原理

总结

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

关于springboot:用这4招优雅的实现Spring-Boot-异步线程间数据传递

1. 手动设置

2. 线程池设置TaskDecorator

3. InheritableThreadLocal

4. TransmittableThreadLocal

TransmittableThreadLocal原理

总结

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复