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

用這4招優雅的實作Spring Boot 異步線程間資料傳遞

1. 手動設定

2. 線程池設定TaskDecorator

3. InheritableThreadLocal

4. TransmittableThreadLocal

TransmittableThreadLocal原理

總結

繼續閱讀

基于Springboot+VUE+Mysql的大學生校園疫情防控管理系統

Using MessageConverter

馴服爛代碼_C＃中的馴服配置

springboot使用 swagger-ui 2.10.5 有關版本更新帶來的問題

Spring Cloud整合Sleuth，當請求完成後，Zipkin沒有鍊路資訊

JAVA自學之路——馬士兵

Rabbitmq指令行手動建立隊列

整合 Spring 開發架構及服務端技術

SSM整合02---Spring-SpringMVC層一、添加web架構支援二、web.xml三、spring-mvc.xml

SSM架構（二）------------表現層的SpringMVC

Intellij idea基于Maven 的SSM 架構整合

Spring資料和Redis

SpringMVC的啟動過程源碼分析1 SpringMVC 配置XML的方式啟動2 SpringMVC純注解模式啟動

Java Spring 架構

spring boot @SpringBootApplication @EnableAutoConfiguration 的工作原理

SpringCloud微服務（四）之ribbon筆記ribbon是什麼？入門案例自定義均衡負載算法

用這4招 優雅的實作Spring Boot 異步線程間資料傳遞

1. 手動設定

2. 線程池設定TaskDecorator

3. InheritableThreadLocal

4. TransmittableThreadLocal

TransmittableThreadLocal原理

總結

繼續閱讀

用這4招優雅的實作Spring Boot 異步線程間資料傳遞