Java中使用阿里巴巴開源庫TransmittableThreadLocal跨線程上下文傳遞講解問題背景Java 的ThreadLocal只能在當(dāng)前線程內(nèi)讀寫數(shù)據(jù)。當(dāng)任務(wù)被提交到線程池執(zhí)行時(shí)子線程線程池中的工作線程無法讀取到父線程設(shè)置的 ThreadLocal 值。ThreadLocalStringcontextnewThreadLocal();context.set(user-007);executorService.submit(()-{System.out.println(context.get());// 輸出 null});JDK 提供了InheritableThreadLocal可以在new Thread()創(chuàng)建子線程時(shí)復(fù)制父線程的值。但線程池中的線程是復(fù)用的不是每次都新建的所以InheritableThreadLocal在線程池場(chǎng)景下也失效。注博客https://blog.csdn.net/badao_liumang_qizhiTransmittableThreadLocal 是什么TransmittableThreadLocalTTL是阿里巴巴開源的一個(gè)庫解決的核心問題是在線程池復(fù)用線程時(shí)仍然能正確傳遞 ThreadLocal 值。原理簡(jiǎn)述在任務(wù)提交到線程池時(shí)捕獲當(dāng)前線程的 TTL 值快照在任務(wù)執(zhí)行前回放快照到工作線程任務(wù)執(zhí)行完畢后恢復(fù)工作線程原來的值。提交線程父 線程池工作線程子 ┌──────────────┐ ┌──────────────┐ │ TTL 007 │ │ TTL null │ │ │ │ │ │ submit(task) ├───┐ │ │ └──────────────┘ │ └──────────────┘ │ │ ① 提交時(shí)捕獲快照 {007} │ ② 執(zhí)行前回放 → TTL 007 │ ③ 執(zhí)行中task 讀到 007 │ ④ 執(zhí)行后恢復(fù) → TTL nullMaven 依賴dependencygroupIdcom.alibaba/groupIdartifactIdtransmittable-thread-local/artifactIdversion2.14.3/version/dependency基礎(chǔ)用法1. 聲明 TransmittableThreadLocalimportcom.alibaba.ttl.TransmittableThreadLocal;publicclassRequestContext{privatestaticfinalTransmittableThreadLocalStringUSER_IDnewTransmittableThreadLocal();privatestaticfinalTransmittableThreadLocalStringTRACE_IDnewTransmittableThreadLocal();publicstaticvoidsetUserId(StringuserId){USER_ID.set(userId);}publicstaticStringgetUserId(){returnUSER_ID.get();}publicstaticvoidsetTraceId(StringtraceId){TRACE_ID.set(traceId);}publicstaticStringgetTraceId(){returnTRACE_ID.get();}publicstaticvoidclear(){USER_ID.remove();TRACE_ID.remove();}}2. 包裝線程池TTL 需要對(duì)線程池進(jìn)行包裝才能生效有三種方式importcom.alibaba.ttl.threadpool.TtlExecutors;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;// 方式一包裝已有線程池推薦最常用ExecutorServicerawPoolExecutors.newFixedThreadPool(4);ExecutorServicettlPoolTtlExecutors.getTtlExecutorService(rawPool);// 方式二包裝 ScheduledExecutorServiceScheduledExecutorServicerawScheduledExecutors.newScheduledThreadPool(2);ScheduledExecutorServicettlScheduledTtlExecutors.getTtlScheduledExecutorService(rawScheduled);3. 使用示例publicclassDemo{privatestaticfinalExecutorServicepoolTtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(4));publicstaticvoidmain(String[]args)throwsException{// 主線程設(shè)置上下文RequestContext.setUserId(007);RequestContext.setTraceId(trace-abc-123);// 提交到線程池子線程能讀到pool.submit(()-{System.out.println(RequestContext.getUserId());// 007System.out.println(RequestContext.getTraceId());// trace-abc-123}).get();// 修改后再提交子線程讀到新值RequestContext.setUserId(008);pool.submit(()-{System.out.println(RequestContext.getUserId());// 008}).get();RequestContext.clear();}}三種傳遞方式對(duì)比方式使用場(chǎng)景說明包裝線程池推薦首選TtlExecutors.getTtlExecutorService(pool)包裝 Runnable/Callable無法修改線程池時(shí)TtlRunnable.get(runnable)Java Agent無侵入全局生效JVM 啟動(dòng)參數(shù)-javaagent:transmittable-thread-local-x.x.x.jar包裝 Runnable不改線程池importcom.alibaba.ttl.TtlRunnable;ExecutorServicerawPoolExecutors.newFixedThreadPool(4);RequestContext.setUserId(007);// 包裝任務(wù)而非線程池Runnabletask()-System.out.println(RequestContext.getUserId());RunnablettlTaskTtlRunnable.get(task);rawPool.submit(ttlTask);// 輸出 007Java Agent 方式零侵入JVM 啟動(dòng)參數(shù)加上-javaagent:/path/to/transmittable-thread-local-2.14.3.jar無需包裝線程池或 Runnable所有 ThreadLocal 自動(dòng)具備跨線程傳遞能力。適合無法修改代碼的場(chǎng)景或全局啟用。Spring Boot 集成配置異步線程池importcom.alibaba.ttl.threadpool.TtlExecutors;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;importjava.util.concurrent.Executor;ConfigurationpublicclassThreadPoolConfig{Bean(asyncExecutor)publicExecutorasyncExecutor(){ThreadPoolTaskExecutorexecutornewThreadPoolTaskExecutor();executor.setCorePoolSize(8);executor.setMaxPoolSize(16);executor.setQueueCapacity(200);executor.setThreadNamePrefix(async-);executor.initialize();// 關(guān)鍵用 TTL 包裝returnTtlExecutors.getTtlExecutorService(executor.getThreadPoolExecutor());}}攔截器設(shè)置上下文importjakarta.servlet.http.HttpServletRequest;importjakarta.servlet.http.HttpServletResponse;importorg.springframework.stereotype.Component;importorg.springframework.web.servlet.HandlerInterceptor;ComponentpublicclassContextInterceptorimplementsHandlerInterceptor{OverridepublicbooleanpreHandle(HttpServletRequestrequest,HttpServletResponseresponse,Objecthandler){RequestContext.setUserId(request.getHeader(X-User-Id));RequestContext.setTraceId(request.getHeader(X-Trace-Id));returntrue;}OverridepublicvoidafterCompletion(HttpServletRequestrequest,HttpServletResponseresponse,Objecthandler,Exceptionex){RequestContext.clear();// 必須清理防止內(nèi)存泄漏}}業(yè)務(wù)使用ServicepublicclassOrderService{AutowiredQualifier(asyncExecutor)privateExecutorasyncExecutor;publicvoidcreateOrder(OrderDTOdto){// 當(dāng)前HTTP線程設(shè)置的上下文// RequestContext.getUserId() 007// 異步處理子線程也能讀到 007CompletableFuture.runAsync(()-{StringuserIdRequestContext.getUserId();// 007 ?auditService.record(userId,創(chuàng)建訂單);},asyncExecutor);}}與 InheritableThreadLocal 的區(qū)別// 測(cè)試代碼InheritableThreadLocalStringitlnewInheritableThreadLocal();TransmittableThreadLocalStringttlnewTransmittableThreadLocal();ExecutorServicepoolExecutors.newFixedThreadPool(1);// 只有1個(gè)線程必定復(fù)用// 第一次提交線程首次創(chuàng)建ITL 和 TTL 都能傳遞itl.set(A);ttl.set(A);pool.submit(()-{System.out.println(ITLitl.get());// ASystem.out.println(TTLttl.get());// A}).get();// 第二次提交線程被復(fù)用ITL 還是舊值TTL 是新值itl.set(B);ttl.set(B);pool.submit(()-{System.out.println(ITLitl.get());// A舊值線程復(fù)用不會(huì)重新繼承System.out.println(TTLttl.get());// B正確每次提交重新捕獲}).get();特性ThreadLocalInheritableThreadLocalTransmittableThreadLocal當(dāng)前線程讀寫???new Thread 繼承???線程池復(fù)用傳遞???值更新后子線程感知-??適用場(chǎng)景場(chǎng)景說明鏈路追蹤traceId、spanId 跨線程透?jìng)饔脩羯矸輦鬟fHTTP 攔截器設(shè)置 userId異步任務(wù)中使用日志 MDC跨線程保持日志上下文字段一致多租戶隔離tenantId 在線程池中不丟失灰度/AB測(cè)試流量標(biāo)記在異步鏈路中傳遞注意事項(xiàng)必須清理HTTP 請(qǐng)求結(jié)束后調(diào)remove()防止線程池復(fù)用時(shí)數(shù)據(jù)泄漏到下一個(gè)請(qǐng)求不能跨 JVMTTL 只解決進(jìn)程內(nèi)跨線程問題跨 MQ、RPC 調(diào)用需要通過消息 Header 或 RPC Context 手動(dòng)傳遞包裝要完整項(xiàng)目中所有線程池都需要包裝遺漏任何一個(gè)都會(huì)導(dǎo)致該路徑上下文丟失避免存大對(duì)象TTL 每次提交任務(wù)都會(huì)做快照拷貝淺拷貝存大對(duì)象有性能和一致性風(fēng)險(xiǎn)配合 RPC 框架Dubbo、gRPC 等通常有自己的 Context 傳遞機(jī)制TTL 負(fù)責(zé)進(jìn)程內(nèi)RPC Context 負(fù)責(zé)進(jìn)程間TTL 不能解決的問題進(jìn)程A (線程池內(nèi)傳遞 ?) 進(jìn)程B (需要額外手段) ┌─────────────────────┐ ┌─────────────────────┐ │ HTTP線程 → 線程池 │── MQ/RPC ──→│ 消費(fèi)線程 │ │ TTL 有效 ? │ │ TTL 管不到 ? │ └─────────────────────┘ └─────────────────────┘ 需要通過 MQ Header / RPC Attachment 傳遞跨進(jìn)程傳遞需要配合其他手段MQ UserProperty、RPC Attachment、HTTP Header 等TTL 只負(fù)責(zé)進(jìn)程內(nèi)線程間的透?jìng)鳌? Java中使用阿里巴巴開源庫TransmittableThreadLocal：跨線程上下文傳遞講解
Java中使用阿里巴巴開源庫TransmittableThreadLocal跨線程上下文傳遞講解問題背景Java 的ThreadLocal只能在當(dāng)前線程內(nèi)讀寫數(shù)據(jù)。當(dāng)任務(wù)被提交到線程池執(zhí)行時(shí)子線程線程池中的工作線程無法讀取到父線程設(shè)置的 ThreadLocal 值。ThreadLocalStringcontextnewThreadLocal();context.set(user-007);executorService.submit(()-{System.out.println(context.get());// 輸出 null});JDK 提供了InheritableThreadLocal可以在new Thread()創(chuàng)建子線程時(shí)復(fù)制父線程的值。但線程池中的線程是復(fù)用的不是每次都新建的所以InheritableThreadLocal在線程池場(chǎng)景下也失效。注博客https://blog.csdn.net/badao_liumang_qizhiTransmittableThreadLocal 是什么TransmittableThreadLocalTTL是阿里巴巴開源的一個(gè)庫解決的核心問題是在線程池復(fù)用線程時(shí)仍然能正確傳遞 ThreadLocal 值。原理簡(jiǎn)述在任務(wù)提交到線程池時(shí)捕獲當(dāng)前線程的 TTL 值快照在任務(wù)執(zhí)行前回放快照到工作線程任務(wù)執(zhí)行完畢后恢復(fù)工作線程原來的值。提交線程父 線程池工作線程子 ┌──────────────┐ ┌──────────────┐ │ TTL 007 │ │ TTL null │ │ │ │ │ │ submit(task) ├───┐ │ │ └──────────────┘ │ └──────────────┘ │ │ ① 提交時(shí)捕獲快照 {007} │ ② 執(zhí)行前回放 → TTL 007 │ ③ 執(zhí)行中task 讀到 007 │ ④ 執(zhí)行后恢復(fù) → TTL nullMaven 依賴dependencygroupIdcom.alibaba/groupIdartifactIdtransmittable-thread-local/artifactIdversion2.14.3/version/dependency基礎(chǔ)用法1. 聲明 TransmittableThreadLocalimportcom.alibaba.ttl.TransmittableThreadLocal;publicclassRequestContext{privatestaticfinalTransmittableThreadLocalStringUSER_IDnewTransmittableThreadLocal();privatestaticfinalTransmittableThreadLocalStringTRACE_IDnewTransmittableThreadLocal();publicstaticvoidsetUserId(StringuserId){USER_ID.set(userId);}publicstaticStringgetUserId(){returnUSER_ID.get();}publicstaticvoidsetTraceId(StringtraceId){TRACE_ID.set(traceId);}publicstaticStringgetTraceId(){returnTRACE_ID.get();}publicstaticvoidclear(){USER_ID.remove();TRACE_ID.remove();}}2. 包裝線程池TTL 需要對(duì)線程池進(jìn)行包裝才能生效有三種方式importcom.alibaba.ttl.threadpool.TtlExecutors;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;// 方式一包裝已有線程池推薦最常用ExecutorServicerawPoolExecutors.newFixedThreadPool(4);ExecutorServicettlPoolTtlExecutors.getTtlExecutorService(rawPool);// 方式二包裝 ScheduledExecutorServiceScheduledExecutorServicerawScheduledExecutors.newScheduledThreadPool(2);ScheduledExecutorServicettlScheduledTtlExecutors.getTtlScheduledExecutorService(rawScheduled);3. 使用示例publicclassDemo{privatestaticfinalExecutorServicepoolTtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(4));publicstaticvoidmain(String[]args)throwsException{// 主線程設(shè)置上下文RequestContext.setUserId(007);RequestContext.setTraceId(trace-abc-123);// 提交到線程池子線程能讀到pool.submit(()-{System.out.println(RequestContext.getUserId());// 007System.out.println(RequestContext.getTraceId());// trace-abc-123}).get();// 修改后再提交子線程讀到新值RequestContext.setUserId(008);pool.submit(()-{System.out.println(RequestContext.getUserId());// 008}).get();RequestContext.clear();}}三種傳遞方式對(duì)比方式使用場(chǎng)景說明包裝線程池推薦首選TtlExecutors.getTtlExecutorService(pool)包裝 Runnable/Callable無法修改線程池時(shí)TtlRunnable.get(runnable)Java Agent無侵入全局生效JVM 啟動(dòng)參數(shù)-javaagent:transmittable-thread-local-x.x.x.jar包裝 Runnable不改線程池importcom.alibaba.ttl.TtlRunnable;ExecutorServicerawPoolExecutors.newFixedThreadPool(4);RequestContext.setUserId(007);// 包裝任務(wù)而非線程池Runnabletask()-System.out.println(RequestContext.getUserId());RunnablettlTaskTtlRunnable.get(task);rawPool.submit(ttlTask);// 輸出 007Java Agent 方式零侵入JVM 啟動(dòng)參數(shù)加上-javaagent:/path/to/transmittable-thread-local-2.14.3.jar無需包裝線程池或 Runnable所有 ThreadLocal 自動(dòng)具備跨線程傳遞能力。適合無法修改代碼的場(chǎng)景或全局啟用。Spring Boot 集成配置異步線程池importcom.alibaba.ttl.threadpool.TtlExecutors;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;importjava.util.concurrent.Executor;ConfigurationpublicclassThreadPoolConfig{Bean(asyncExecutor)publicExecutorasyncExecutor(){ThreadPoolTaskExecutorexecutornewThreadPoolTaskExecutor();executor.setCorePoolSize(8);executor.setMaxPoolSize(16);executor.setQueueCapacity(200);executor.setThreadNamePrefix(async-);executor.initialize();// 關(guān)鍵用 TTL 包裝returnTtlExecutors.getTtlExecutorService(executor.getThreadPoolExecutor());}}攔截器設(shè)置上下文importjakarta.servlet.http.HttpServletRequest;importjakarta.servlet.http.HttpServletResponse;importorg.springframework.stereotype.Component;importorg.springframework.web.servlet.HandlerInterceptor;ComponentpublicclassContextInterceptorimplementsHandlerInterceptor{OverridepublicbooleanpreHandle(HttpServletRequestrequest,HttpServletResponseresponse,Objecthandler){RequestContext.setUserId(request.getHeader(X-User-Id));RequestContext.setTraceId(request.getHeader(X-Trace-Id));returntrue;}OverridepublicvoidafterCompletion(HttpServletRequestrequest,HttpServletResponseresponse,Objecthandler,Exceptionex){RequestContext.clear();// 必須清理防止內(nèi)存泄漏}}業(yè)務(wù)使用ServicepublicclassOrderService{AutowiredQualifier(asyncExecutor)privateExecutorasyncExecutor;publicvoidcreateOrder(OrderDTOdto){// 當(dāng)前HTTP線程設(shè)置的上下文// RequestContext.getUserId() 007// 異步處理子線程也能讀到 007CompletableFuture.runAsync(()-{StringuserIdRequestContext.getUserId();// 007 ?auditService.record(userId,創(chuàng)建訂單);},asyncExecutor);}}與 InheritableThreadLocal 的區(qū)別// 測(cè)試代碼InheritableThreadLocalStringitlnewInheritableThreadLocal();TransmittableThreadLocalStringttlnewTransmittableThreadLocal();ExecutorServicepoolExecutors.newFixedThreadPool(1);// 只有1個(gè)線程必定復(fù)用// 第一次提交線程首次創(chuàng)建ITL 和 TTL 都能傳遞itl.set(A);ttl.set(A);pool.submit(()-{System.out.println(ITLitl.get());// ASystem.out.println(TTLttl.get());// A}).get();// 第二次提交線程被復(fù)用ITL 還是舊值TTL 是新值itl.set(B);ttl.set(B);pool.submit(()-{System.out.println(ITLitl.get());// A舊值線程復(fù)用不會(huì)重新繼承System.out.println(TTLttl.get());// B正確每次提交重新捕獲}).get();特性ThreadLocalInheritableThreadLocalTransmittableThreadLocal當(dāng)前線程讀寫???new Thread 繼承???線程池復(fù)用傳遞???值更新后子線程感知-??適用場(chǎng)景場(chǎng)景說明鏈路追蹤traceId、spanId 跨線程透?jìng)饔脩羯矸輦鬟fHTTP 攔截器設(shè)置 userId異步任務(wù)中使用日志 MDC跨線程保持日志上下文字段一致多租戶隔離tenantId 在線程池中不丟失灰度/AB測(cè)試流量標(biāo)記在異步鏈路中傳遞注意事項(xiàng)必須清理HTTP 請(qǐng)求結(jié)束后調(diào)remove()防止線程池復(fù)用時(shí)數(shù)據(jù)泄漏到下一個(gè)請(qǐng)求不能跨 JVMTTL 只解決進(jìn)程內(nèi)跨線程問題跨 MQ、RPC 調(diào)用需要通過消息 Header 或 RPC Context 手動(dòng)傳遞包裝要完整項(xiàng)目中所有線程池都需要包裝遺漏任何一個(gè)都會(huì)導(dǎo)致該路徑上下文丟失避免存大對(duì)象TTL 每次提交任務(wù)都會(huì)做快照拷貝淺拷貝存大對(duì)象有性能和一致性風(fēng)險(xiǎn)配合 RPC 框架Dubbo、gRPC 等通常有自己的 Context 傳遞機(jī)制TTL 負(fù)責(zé)進(jìn)程內(nèi)RPC Context 負(fù)責(zé)進(jìn)程間TTL 不能解決的問題進(jìn)程A (線程池內(nèi)傳遞 ?) 進(jìn)程B (需要額外手段) ┌─────────────────────┐ ┌─────────────────────┐ │ HTTP線程 → 線程池 │── MQ/RPC ──→│ 消費(fèi)線程 │ │ TTL 有效 ? │ │ TTL 管不到 ? │ └─────────────────────┘ └─────────────────────┘ 需要通過 MQ Header / RPC Attachment 傳遞跨進(jìn)程傳遞需要配合其他手段MQ UserProperty、RPC Attachment、HTTP Header 等TTL 只負(fù)責(zé)進(jìn)程內(nèi)線程間的透?jìng)鳌?