spring?Cloud微服務(wù)阿里開源TTL身份信息的線程間復(fù)用

更新時(shí)間：2023年01月06日 15:07:37 作者：碼猿技術(shù)專欄

這篇文章主要為大家介紹了spring?Cloud微服務(wù)中使用阿里開源TTL身份信息的線程間復(fù)用,有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

引言

前面在介紹分布式鏈路追蹤時(shí)講過異步調(diào)用會(huì)丟失鏈路信息，最終的解決方案是使用對(duì)應(yīng)的包裝類重新包裝一下，如下：

RunnableWrapper

CallableWrapper

SupplierWrapper

還有openFeign異步請(qǐng)求丟失上文的問題，這些問題追根究底都是ThreadLocal惹得禍。

由于ThreadLocal只能保存當(dāng)前線程的信息，不能實(shí)現(xiàn)父子線程的繼承。

說到這，很多人想到了InheritableThreadLocal，確實(shí)InheritableThreadLocal能夠?qū)崿F(xiàn)父子線程間傳遞本地變量，但是.....

但是你的程序如果采用線程池，則存在著線程復(fù)用的情況，這時(shí)就不一定能夠?qū)崿F(xiàn)父子線程間傳遞了，因?yàn)樵诰€程在線程池中的存在不是每次使用都會(huì)進(jìn)行創(chuàng)建，InheritableThreadlocal是在線程初始化時(shí)intertableThreadLocals=true才會(huì)進(jìn)行拷貝傳遞。

所以若本次使用的子線程是已經(jīng)被池化的線程，從線程池中取出線下進(jìn)行使用，是沒有經(jīng)過初始化的過程，也就不會(huì)進(jìn)行父子線程的本地變量拷貝。

由于在日常應(yīng)用場(chǎng)景中，絕大多數(shù)都是會(huì)采用線程池的方式進(jìn)行資源的有效管理。

今天就來聊一聊阿里的ThansmittableThreadLocal是如何解決線程池中父子線程本地變量傳遞。

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

InheritableThreadLocal 的問題

在介紹ThansmittableThreadLocal之前先來看一下InheritableThreadLocal 在線程池中的問題，如下代碼：

@Test
public?void?test()?throws?Exception?{
????//單一線程池
????ExecutorService?executorService?=?Executors.newSingleThreadExecutor();
????//InheritableThreadLocal存儲(chǔ)
????InheritableThreadLocal<String>?username?=?new?InheritableThreadLocal<>();
????for?(int?i?=?0;?i?<?10;?i++)?{
????username.set("公眾號(hào)：腳本之家—"+i);
????Thread.sleep(3000);
????CompletableFuture.runAsync(()->?System.out.println(username.get()),executorService);
???}
}

上述代碼中創(chuàng)建了一個(gè)單一線程池，循環(huán)異步調(diào)用，打印一下username，由于核心線程數(shù)是1，勢(shì)必存在線程的復(fù)用。

打印信息如下：

公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—0

看到了嗎？這里并沒有實(shí)現(xiàn)父子線程間的變量傳遞，這也就是InheritableThreadLocal 的局限性。

TransmittableThreadLocal 使用

TransmittableThreadLocal(TTL)：在使用線程池等會(huì)池化復(fù)用線程的執(zhí)行組件情況下，提供ThreadLocal值的傳遞功能，解決異步執(zhí)行時(shí)上下文傳遞的問題。

整個(gè)TransmittableThreadLocal庫的核心功能（用戶API與框架/中間件的集成API、線程池ExecutorService/ForkJoinPool/TimerTask及其線程工廠的Wrapper）。

需求場(chǎng)景：

分布式跟蹤系統(tǒng) 或全鏈路壓測(cè)（即鏈路打標(biāo)）
日志收集記錄系統(tǒng)上下文

官網(wǎng)地址：https://github.com/alibaba/transmittable-thread-local

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

首選需要引入對(duì)應(yīng)的依賴，如下：

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

改造后的代碼如下：

@Test
public?void?test()?throws?Exception?{
????//單一線程池
????ExecutorService?executorService?=?Executors.newSingleThreadExecutor();
????//需要使用TtlExecutors對(duì)線程池包裝一下
????executorService=TtlExecutors.getTtlExecutorService(executorService);
????//TransmittableThreadLocal創(chuàng)建
????TransmittableThreadLocal<String>?username?=?new?TransmittableThreadLocal<>();
????for?(int?i?=?0;?i?<?10;?i++)?{
????username.set("公眾號(hào)：https://github.com/alibaba/transmittable-thread-local—"+i);
????Thread.sleep(3000);
????CompletableFuture.runAsync(()->?System.out.println(username.get()),executorService);
??}
}

需要注意的是需要使用TtlExecutors對(duì)線程池進(jìn)行包裝，代碼如下：

executorService=TtlExecutors.getTtlExecutorService(executorService);

運(yùn)行效果如下：

公眾號(hào)：腳本之家—0
公眾號(hào)：腳本之家—1
公眾號(hào)：腳本之家—2
公眾號(hào)：腳本之家—3
公眾號(hào)：腳本之家—4
公眾號(hào)：腳本之家—5
公眾號(hào)：腳本之家—6
公眾號(hào)：腳本之家—7
公眾號(hào)：腳本之家—8
公眾號(hào)：腳本之家—9

可以看到已經(jīng)能夠?qū)崿F(xiàn)了線程池中的父子線程的數(shù)據(jù)傳遞。

在每次調(diào)用任務(wù)的時(shí)，都會(huì)將當(dāng)前的主線程的TTL數(shù)據(jù)copy到子線程里面，執(zhí)行完成后，再清除掉。同時(shí)子線程里面的修改回到主線程時(shí)其實(shí)并沒有生效。這樣可以保證每次任務(wù)執(zhí)行的時(shí)候都是互不干涉。

簡單應(yīng)用

在 Spring Security 往往需要存儲(chǔ)用戶登錄的詳細(xì)信息，這樣在業(yè)務(wù)方法中能夠隨時(shí)獲取用戶的信息。

在前面的Spring Cloud Gateway整合OAuth2.0實(shí)現(xiàn)統(tǒng)一認(rèn)證鑒權(quán) 文章中筆者是將用戶信息直接存儲(chǔ)在Request中，這樣每次請(qǐng)求都能獲取到對(duì)應(yīng)的信息。

其實(shí)Request中的信息存儲(chǔ)也是通過ThreadLocal完成的，在異步執(zhí)行的時(shí)候還是需要重新轉(zhuǎn)存，這樣一來代碼就變得復(fù)雜。

那么了解了TransmittableThreadLocal 之后，完全可以使用這個(gè)存儲(chǔ)用戶的登錄信息，實(shí)現(xiàn)如下：

/**
?*?@description?使用TransmittableThreadLocal存儲(chǔ)用戶身份信息LoginVal
?*/
public?class?SecurityContextHolder?{
????//使用TTL存儲(chǔ)身份信息
????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中的一次請(qǐng)求對(duì)應(yīng)一個(gè)線程，因此只需要在攔截器中的設(shè)置和移除TransmittableThreadLocal中的信息，代碼如下：

/**
?*?@description?攔截器，在preHandle中解析請(qǐng)求頭的中的token信息，將其放入SecurityContextHolder中
?*??????????????????????在afterCompletion方法中移除對(duì)應(yīng)的ThreadLocal中信息
?*??????????????????????確保每個(gè)請(qǐng)求的用戶信息獨(dú)立
?*/
@Component
public?class?AuthInterceptor?implements?AsyncHandlerInterceptor?{
????/**
?????*?在執(zhí)行controller方法之前將請(qǐng)求頭中的token信息解析出來，放入SecurityContextHolder中（TransmittableThreadLocal）
?????*/
????@Override
????public?boolean?preHandle(HttpServletRequest?request,?HttpServletResponse?response,?Object?handler)?{
????????if?(!(handler?instanceof?HandlerMethod))
????????????return?true;
????????//獲取請(qǐng)求頭中的加密的用戶信息
????????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);
????????//封裝數(shù)據(jù)到ThreadLocal中
????????SecurityContextHolder.set(loginVal);
????????return?true;
????}
????/**
?????*?在視圖渲染之后執(zhí)行，意味著一次請(qǐng)求結(jié)束，清除TTL中的身份信息
?????*/
????@Override
????public?void?afterCompletion(HttpServletRequest?request,?HttpServletResponse?response,?Object?handler,?Exception?ex){
????????SecurityContextHolder.remove();
????}
}

原理

從定義來看，TransimittableThreadLocal繼承于InheritableThreadLocal，并實(shí)現(xiàn)TtlCopier接口，它里面只有一個(gè)copy方法。所以主要是對(duì)InheritableThreadLocal的擴(kuò)展。

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

在TransimittableThreadLocal中添加holder屬性。這個(gè)屬性的作用就是被標(biāo)記為具備線程傳遞資格的對(duì)象都會(huì)被添加到這個(gè)對(duì)象中。

要標(biāo)記一個(gè)類，比較容易想到的方式，就是給這個(gè)類新增一個(gè)Type字段，還有一個(gè)方法就是將具備這種類型的的對(duì)象都添加到一個(gè)靜態(tài)全局集合中。之后使用時(shí)，這個(gè)集合里的所有值都具備這個(gè)標(biāo)記。

//?1.?holder本身是一個(gè)InheritableThreadLocal對(duì)象
//?2.?這個(gè)holder對(duì)象的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方法，實(shí)現(xiàn)上直接將父線程的屬性作為子線程的本地變量對(duì)象。
???*/
??@Override
??protected?WeakHashMap<TransmittableThreadLocal<Object>,??>?childValue(WeakHashMap<TransmittableThreadLocal<Object>,??>?parentValue)?{
????return?new?WeakHashMap<TransmittableThreadLocal<Object>,?Object>(parentValue);
??}
};

應(yīng)用代碼是通過TtlExecutors工具類對(duì)線程池對(duì)象進(jìn)行包裝。工具類只是簡單的判斷，輸入的線程池是否已經(jīng)被包裝過、非空校驗(yàn)等，然后返回包裝類ExecutorServiceTtlWrapper。根據(jù)不同的線程池類型，有不同和的包裝類。

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

進(jìn)入包裝類ExecutorServiceTtlWrapper?？梢宰⒁獾讲徽撌峭ㄟ^ExecutorServiceTtlWrapper#submit方法或者是ExecutorTtlWrapper#execute方法，都會(huì)將線程對(duì)象包裝成TtlCallable或者TtlRunnable，用于在真正執(zhí)行run方法前做一些業(yè)務(wù)邏輯。

/**
?*?在ExecutorServiceTtlWrapper實(shí)現(xiàn)submit方法
?*/
@NonNull
@Override
public?<T>?Future<T>?submit(@NonNull?Callable<T>?task)?{
??return?executorService.submit(TtlCallable.get(task));
}
/**
?*?在ExecutorTtlWrapper實(shí)現(xiàn)execute方法
?*/
@Override
public?void?execute(@NonNull?Runnable?command)?{
??executor.execute(TtlRunnable.get(command));
}

所以，重點(diǎn)的核心邏輯應(yīng)該是在TtlCallable#call()或者TtlRunnable#run()中。以下以TtlCallable為例，TtlRunnable同理類似。在分析call()方法之前，先看一個(gè)類Transmitter

public?static?class?Transmitter?{
??/**
????*?捕獲當(dāng)前線程中的是所有TransimittableThreadLocal和注冊(cè)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;
??}
??/**
????*?捕獲注冊(cè)的ThreadLocal的值,也就是原本線程中的ThreadLocal,可以注冊(cè)到TTL中，在
????*?進(jìn)行線程池本地變量傳遞時(shí)也會(huì)被傳遞。
????*/
??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;
??}
??/**
????*?將捕獲到的本地變量進(jìn)行替換子線程的本地變量，并且返回子線程現(xiàn)有的本地變量副本backup。
????*?用于在執(zhí)行run/call方法之后，將本地變量副本恢復(fù)。
????*/
??@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)?{
????//?創(chuàng)建副本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();
??????//?對(duì)當(dāng)前線程的本地變量進(jìn)行副本拷貝
??????backup.put(threadLocal,?threadLocal.get());
??????//?若出現(xiàn)調(diào)用線程中不存在某個(gè)線程變量，而線程池中線程有，則刪除線程池中對(duì)應(yīng)的本地變量
??????if?(!captured.containsKey(threadLocal))?{
????????iterator.remove();
????????threadLocal.superRemove();
??????}
????}
????//?將捕獲的TTL值打入線程池獲取到的線程TTL中。
????setTtlValuesTo(captured);
????//?是一個(gè)擴(kuò)展點(diǎn)，調(diào)用TTL的beforeExecute方法。默認(rèn)實(shí)現(xiàn)為空
????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)?{
????//?擴(kuò)展點(diǎn)，調(diào)用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();
??????}
????}
????//?將本地變量恢復(fù)成備份版本
????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;
????}
??}

進(jìn)入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!");
??}
??//?調(diào)用replay方法將捕獲到的當(dāng)前線程的本地變量，傳遞給線程池線程的本地變量，
??//?并且獲取到線程池線程覆蓋之前的本地變量副本。
??Object?backup?=?replay(captured);
??try?{
????//?線程方法調(diào)用
????return?callable.call();
??}?finally?{
????//?使用副本進(jìn)行恢復(fù)。
????restore(backup);
??}
}

到這基本上線程池方式傳遞本地變量的核心代碼已經(jīng)大概看完了?？偟膩碚f在創(chuàng)建TtlCallable對(duì)象是，調(diào)用capture()方法捕獲調(diào)用方的本地線程變量，在call()執(zhí)行時(shí)，將捕獲到的線程變量，替換到線程池所對(duì)應(yīng)獲取到的線程的本地變量中，并且在執(zhí)行完成之后，將其本地變量恢復(fù)到調(diào)用之前。