【Java实战】低侵入的线程池值传递

欢迎来到啾啾的博客🐱。
记录学习点滴。分享工作思考和实用技巧，偶尔也分享一些杂谈💬。
有很多很多不足的地方，欢迎评论交流，感谢您的阅读和评论😄。

引言

在之前的Java基础ThreadLocal篇章中，我们有了解到，ThreadLocal存储值线程安全的本质，是获取线程实例独享的的ThreadLocalMap属性。且k-v内容为this-value。在线程池复用线程的场景中，如果每次使用ThreadLocal存储值而不清除，线程的ThreadLocalMap将会持续扩容，直至内存溢出。
为此，我们需要在每次使用完ThreadLocal后进行remove操作。

但是在复杂场景中，我们可能忘记清理。
甚至在一些场景，我们希望主线程中ThreadLocal的值可以低侵入地传递到子线程中，比如如用于追踪请求调用链路的TraceID。

那么，我们应该怎么做呢？

InheritableThreadLocal

标准 ThreadLocal 的值不会自动从父线程传递到子线程。为此，Java 提供了 InheritableThreadLocal。
当你创建一个新线程时，子线程会自动继承父线程中 InheritableThreadLocal 变量的值。

public class TraceContext {private static final InheritableThreadLocal<String> TRACE_ID_HOLDER = new InheritableThreadLocal<>();public static void setTraceId(String traceId) {TRACE_ID_HOLDER.set(traceId);}public static String getTraceId() {return TRACE_ID_HOLDER.get();}public static void clearTraceId() {TRACE_ID_HOLDER.remove();}public static void main(String[] args) {// 主线程TraceContext.setTraceId("Main-thread-trace-id");new Thread(() -> {/*会打印 "main-thread-trace-id"注意：子线程修改不会影响父线程，父线程后续修改也不会影响已创建的子线程*/System.out.println("Child thread traceId: " + TraceContext.getTraceId());TraceContext.setTraceId("child-thread-trace-id");System.out.println("Child thread traceId: " + TraceContext.getTraceId());TraceContext.clearTraceId(); // 最好也在子线程用完后清理}).start();// 主线程等待try {Thread.sleep(1000);System.out.println("Main thread traceId: " + TraceContext.getTraceId());} catch (InterruptedException e) {throw new RuntimeException(e);} finally {// 父线程也需要清理(通常在业务流程结束时清理)TraceContext.clearTraceId();}System.out.println("Main thread traceId: " + TraceContext.getTraceId());}
}

InheritableThreadLocal 的值是在子线程创建时从父线程复制的。
但是当使用是线程池时，InheritableThreadLocal有一个问题。
当一个任务提交给线程池，线程池复用一个已存在的线程时，这个被复用的线程不会从提交任务的当前线程那里重新继承 InheritableThreadLocal 的值。它会保留上一个任务结束时（或者它被创建时）的状态。
这会导致如果如果任务A在线程T1中设置了 InheritableThreadLocal 的值为 valA，任务A结束后没有清理。然后任务B（由不同的请求触发，期望的 InheritableThreadLocal 值为 valB）被分配到同一个线程T1，它会看到 valA 而不是期望的 valB。

InheritableThreadLocal解决了传递的问题，但是并不能解决自动处理的问题，还是不能“能够忘记清理”。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;public class TraceContext {// 使用 InheritableThreadLocalprivate static final InheritableThreadLocal<String> TRACE_ID_HOLDER = new InheritableThreadLocal<>();public static void setTraceId(String traceId) {System.out.println("[" + Thread.currentThread().getName() + "] 设置 TraceID: " + traceId);TRACE_ID_HOLDER.set(traceId);}public static String getTraceId() {String traceId = TRACE_ID_HOLDER.get();// System.out.println("[" + Thread.currentThread().getName() + "] 获取 TraceID: " + traceId); // 频繁打印会比较乱return traceId;}public static void clearTraceId() {System.out.println("[" + Thread.currentThread().getName() + "] 清理 TraceID. 清理前的值: " + TRACE_ID_HOLDER.get());TRACE_ID_HOLDER.remove();}public static void main(String[] args) throws InterruptedException {// 创建一个单线程的线程池，方便观察线程复用ExecutorService executorService = Executors.newSingleThreadExecutor();// ExecutorService executorService = Executors.newFixedThreadPool(1); // 等效System.out.println("===== 场景演示：InheritableThreadLocal 与线程池复用问题 =====");// --- 任务1：由主线程提交 ---// 主线程设置自己的 TraceIDTraceContext.setTraceId("主线程-为任务1设置的TraceID");System.out.println("[" + Thread.currentThread().getName() + "] 准备提交任务1。主线程当前 TraceID: " + TraceContext.getTraceId());executorService.submit(() -> {String threadName = Thread.currentThread().getName();System.out.println("[" + threadName + "] 任务1开始执行。继承到的 TraceID: " + TraceContext.getTraceId());// 任务1 设置自己业务相关的 TraceIDTraceContext.setTraceId("任务1特定的TraceID");System.out.println("[" + threadName + "] 任务1设置自身TraceID后。当前 TraceID: " + TraceContext.getTraceId());// 模拟任务执行try {Thread.sleep(50);} catch (InterruptedException e) {Thread.currentThread().interrupt();}// 关键：任务1在结束前忘记清理自己设置的 TraceIDSystem.out.println("[" + threadName + "] 任务1执行完毕。退出任务前 TraceID: " + TraceContext.getTraceId());// TraceContext.clearTraceId(); // <-- 如果这里调用了清理，任务2就不会看到脏数据});// 等待任务1执行完毕，确保线程已被“污染”Thread.sleep(200); // 确保任务1完成System.out.println("\n[" + Thread.currentThread().getName() + "] 任务1提交后，主线程的 TraceID (应保持不变): " + TraceContext.getTraceId());TraceContext.clearTraceId(); // 主线程清理自己的 TraceID ("主线程-为任务1设置的TraceID")System.out.println("[" + Thread.currentThread().getName() + "] 主线程清理自身TraceID后: " + TraceContext.getTraceId() + "\n");// --- 任务2：同样由主线程提交（此时主线程可能为任务2设置了新的TraceID） ---// 主线程为任务2的上下文设置新的 TraceIDTraceContext.setTraceId("主线程-为任务2设置的TraceID");System.out.println("[" + Thread.currentThread().getName() + "] 准备提交任务2。主线程当前 TraceID: " + TraceContext.getTraceId());executorService.submit(() -> {String threadName = Thread.currentThread().getName();// 问题点：线程池线程（从任务1复用而来）仍然持有 "任务1特定的TraceID"// 它并不会从提交任务2的主线程那里继承 "主线程-为任务2设置的TraceID"System.out.println("[" + threadName + "] 任务2开始执行。继承到/残留的 TraceID: " + TraceContext.getTraceId() + " <<-- 问题点！这是任务1的残留，不是主线程为任务2设的值");// 如果任务2现在设置自己的ID，它会覆盖旧的脏数据TraceContext.setTraceId("任务2特定的TraceID");System.out.println("[" + threadName + "] 任务2设置自身TraceID后。当前 TraceID: " + TraceContext.getTraceId());// 模拟任务执行try {Thread.sleep(50);} catch (InterruptedException e) {Thread.currentThread().interrupt();}// 任务2 正确地清理了它的 TraceIDTraceContext.clearTraceId();System.out.println("[" + threadName + "] 任务2执行完毕。清理后 TraceID: " + TraceContext.getTraceId());});// 等待任务2执行完毕Thread.sleep(200);System.out.println("\n[" + Thread.currentThread().getName() + "] 任务2提交后，主线程的 TraceID (应保持不变): " + TraceContext.getTraceId());TraceContext.clearTraceId(); // 主线程清理自己的 TraceID ("主线程-为任务2设置的TraceID")executorService.shutdown();try {if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {executorService.shutdownNow();}} catch (InterruptedException e) {executorService.shutdownNow();Thread.currentThread().interrupt();}System.out.println("\n[" + Thread.currentThread().getName() + "] 主线程执行完毕。最终 TraceID: " + TraceContext.getTraceId());System.out.println("===== 场景演示结束 =====");}
}

Alibaba TransmittableThreadLocal

为了解决 InheritableThreadLocal 在线程池复用场景下的问题，通常的做法是：

• 在父线程提交任务给线程池之前，获取父线程的 InheritableThreadLocal 上下文。
• 包装 Runnable 或 Callable，使得在任务实际执行前（在线程池线程中），将捕获的上下文设置到当前“线程池线程”的 InheritableThreadLocal 中；任务执行完毕后，再清理掉。

Alibaba TransmittableThreadLocal（TTL）就是这样做的。它通过包装 Runnable, Callable, ExecutorService 等来实现上下文的正确传递和恢复。

import com.alibaba.ttl.TransmittableThreadLocal; // 引入TTL
import com.alibaba.ttl.threadpool.TtlExecutors; // 引入TtlExecutors (推荐方式)import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;public class TraceContextWithTTL {// 将 InheritableThreadLocal 替换为 TransmittableThreadLocalprivate static final TransmittableThreadLocal<String> TRACE_ID_HOLDER = new TransmittableThreadLocal<>();public static void setTraceId(String traceId) {System.out.println("[" + Thread.currentThread().getName() + "] 设置 TraceID: " + traceId);TRACE_ID_HOLDER.set(traceId);}public static String getTraceId() {String traceId = TRACE_ID_HOLDER.get();return traceId;}public static void clearTraceId() {System.out.println("[" + Thread.currentThread().getName() + "] 清理 TraceID. 清理前的值: " + TRACE_ID_HOLDER.get());TRACE_ID_HOLDER.remove();}public static void main(String[] args) throws InterruptedException {// 1. 创建一个普通的线程池ExecutorService originalExecutorService = Executors.newSingleThreadExecutor();// 2. 使用 TtlExecutors 包装原始线程池，使其支持TTL功能// 这样提交给 ttlExecutorService 的 Runnable/Callable 会被自动包装ExecutorService ttlExecutorService = TtlExecutors.getTtlExecutorService(originalExecutorService);System.out.println("===== 场景演示：TransmittableThreadLocal (TTL) 解决线程池复用问题 =====");// --- 任务1：由主线程提交 ---TraceContextWithTTL.setTraceId("主线程-为任务1设置的TraceID");System.out.println("[" + Thread.currentThread().getName() + "] 准备提交任务1。主线程当前 TraceID: " + TraceContextWithTTL.getTraceId());ttlExecutorService.submit(() -> { // 提交给包装后的 ttlExecutorServiceString threadName = Thread.currentThread().getName();// TTL 会确保这里能正确获取到父线程(main)在提交任务时设置的TraceIDSystem.out.println("[" + threadName + "] 任务1开始执行。通过TTL获取到的 TraceID: " + TraceContextWithTTL.getTraceId());TraceContextWithTTL.setTraceId("任务1特定的TraceID");System.out.println("[" + threadName + "] 任务1设置自身TraceID后。当前 TraceID: " + TraceContextWithTTL.getTraceId());try {Thread.sleep(50);} catch (InterruptedException e) {Thread.currentThread().interrupt();}// 任务1 仍然忘记清理 (用于演示TTL的上下文恢复能力)System.out.println("[" + threadName + "] 任务1执行完毕。退出任务前 TraceID: " + TraceContextWithTTL.getTraceId());// 即使这里没有 clearTraceId()，TTL 也会在任务执行后恢复线程池线程的原有TTL状态});Thread.sleep(200); // 确保任务1完成System.out.println("\n[" + Thread.currentThread().getName() + "] 任务1提交后，主线程的 TraceID (应保持不变): " + TraceContextWithTTL.getTraceId());TraceContextWithTTL.clearTraceId();System.out.println("[" + Thread.currentThread().getName() + "] 主线程清理自身TraceID后: " + TraceContextWithTTL.getTraceId() + "\n");// --- 任务2：同样由主线程提交 ---TraceContextWithTTL.setTraceId("主线程-为任务2设置的TraceID");System.out.println("[" + Thread.currentThread().getName() + "] 准备提交任务2。主线程当前 TraceID: " + TraceContextWithTTL.getTraceId());ttlExecutorService.submit(() -> { // 再次提交给包装后的 ttlExecutorServiceString threadName = Thread.currentThread().getName();// 关键点：即使任务1没有清理，TTL 也会确保任务2在开始时，// 其 TransmittableThreadLocal 的值是从提交任务2的父线程(main)那里正确传递过来的。// 不会再看到任务1残留的 "任务1特定的TraceID"。System.out.println("[" + threadName + "] 任务2开始执行。通过TTL获取到的 TraceID: " + TraceContextWithTTL.getTraceId() + " <<-- 正确！这是主线程为任务2设置的值");TraceContextWithTTL.setTraceId("任务2特定的TraceID");System.out.println("[" + threadName + "] 任务2设置自身TraceID后。当前 TraceID: " + TraceContextWithTTL.getTraceId());try {Thread.sleep(50);} catch (InterruptedException e) {Thread.currentThread().interrupt();}TraceContextWithTTL.clearTraceId(); // 任务2 遵循良好实践，进行了清理System.out.println("[" + threadName + "] 任务2执行完毕。清理后 TraceID: " + TraceContextWithTTL.getTraceId());});Thread.sleep(200); // 确保任务2完成System.out.println("\n[" + Thread.currentThread().getName() + "] 任务2提交后，主线程的 TraceID (应保持不变): " + TraceContextWithTTL.getTraceId());TraceContextWithTTL.clearTraceId();// 关闭原始线程池 (TtlExecutors 包装的线程池会委托给原始线程池)originalExecutorService.shutdown();try {if (!originalExecutorService.awaitTermination(5, TimeUnit.SECONDS)) {originalExecutorService.shutdownNow();}} catch (InterruptedException e) {originalExecutorService.shutdownNow();Thread.currentThread().interrupt();}System.out.println("\n[" + Thread.currentThread().getName() + "] 主线程执行完毕。最终 TraceID: " + TraceContextWithTTL.getTraceId());System.out.println("===== 场景演示结束 =====");// 补充：如果不想包装 ExecutorService，也可以手动包装 Runnable/Callable// ExecutorService plainExecutor = Executors.newSingleThreadExecutor();// TraceContextWithTTL.setTraceId("Manually-Wrapped-TraceID");// Runnable originalRunnable = () -> {//     System.out.println("[" + Thread.currentThread().getName() + "] 手动包装的Runnable TraceID: " + TraceContextWithTTL.getTraceId());//     TraceContextWithTTL.clearTraceId();// };// Runnable ttlRunnable = TtlRunnable.get(originalRunnable); // 手动包装// plainExecutor.submit(ttlRunnable);// plainExecutor.shutdown();// plainExecutor.awaitTermination(1, TimeUnit.SECONDS);// TraceContextWithTTL.clearTraceId();}
}