Java 利用TTL统一线程池上下文流水

日志串流水，统一一次请求的流水号。同一个线程里比较简单，线程上下文ThreadLocal直接用。当遇到起异步线程或者线程池的，我们怎么把业务线程的流水传递过去呢？

通常两种实现方式：

封装线程池，即把原始线程池包装一下，覆写execute方法。
线程池原生，封装提交的runnable和callbable任务。

思路都是，把主线程里的ThreadLocal拿出来，再恢复到异步线程里。（lambda表达式内的就表示异步线程里了，外部还是主线程）

TraceId 上下文工具类

使用上下文 ThreadLocal 存储流水号。

public class TraceIdContext {
    // TraceID 上下文 Key
    public static final String TRACE_ID_KEY = "traceId";

    // 线程本地存储 TraceID
    private static final ThreadLocal<String> TRACE_ID_HOLDER = new ThreadLocal<>();
    /**
     * 设置 TraceID 外部传入
     */
    public static void set(String traceId) {
        TRACE_ID_HOLDER.set(traceId);
    }
    /**
     * 获取当前线程 TraceID
     */
    public static String get() {
        return TRACE_ID_HOLDER.get();
    }
    /**
     * 清除 TraceID（防止内存泄漏）
     */
    public static void clear() {
        TRACE_ID_HOLDER.remove();
    }
    /**
     * 捕获当前上下文（用于跨线程传递）
     * 扩展：如需传递更多上下文（如用户ID），可在此添加
     */
    public static Map<String, String> captureContext() {
        Map<String, String> map = new HashMap<>();
        map.put(TRACE_ID_KEY, get());
        return map;
    }

    /**
     * 恢复上下文（异步线程中使用）
     */
    public static void restoreContext(Map<String, String> context) {
        if (context != null && context.containsKey(TRACE_ID_KEY)) {
            set(context.get(TRACE_ID_KEY));
        }
    }
}

自定义线程池包装器

包装线程池的 execute 方法，在提交任务时携带当前线程的 TraceID，异步执行时恢复到新线程中。

实现 Executor 接口
构造方法包装进来原始的线程池
重写executor 方法，上下文恢复

/**
 * 携带 TraceID 的线程池包装器
 */
public class TraceIdExecutorWrapper implements Executor {
    // 原始线程池
    private final Executor delegate;

    /**
     * 通过构造方法注入进来原始线程池
     * @param delegate
     */
    public TraceIdExecutorWrapper(Executor delegate) {
        this.delegate = delegate;
    }

    /**
     * 覆盖execute方法，来包装原始的 runnable任务
     * @param runnable the runnable task
     */
    @Override
    public void execute(Runnable runnable) {
        // 捕获当前线程的 TraceID
        String currentTraceId = TraceIdContext.get();
        // 包装任务，执行时恢复 TraceID
        delegate.execute(() -> {
            try {
                // 将主线程的 TraceID 绑定到异步线程
                TraceIdContext.set(currentTraceId);
                runnable.run(); // 执行原始任务
            } finally {
                // 执行完清除，避免 ThreadLocal 内存泄漏
                TraceIdContext.clear();
            }
        });
    }
}

实例化线程池时覆盖execute方法，匿名类形式

new ThreadPoolExecutor() 后，直接用匿名类形式覆写execute()方法

ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 8,
                60L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(100),
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "my-pool-" + threadNumber.getAndIncrement());
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy()) {
            
            // 匿名类，覆写execute()方法
            
            @Override
            public void execute(@NotNull Runnable command) {
                // 装饰 runnable 逻辑
                Map<String, String> context = TraceIdContext.captureContext();

                Runnable command1 = () -> {
                    try {
                        TraceIdContext.restoreContext(context); // 恢复 TraceID 到异步线程
                        command.run(); // 执行原始任务
                    } finally {
                        TraceIdContext.clear(); // 强制清除，避免 ThreadLocal 泄漏
                    }
                };
                // 装饰后，调用super.execute()
                super.execute(command1);
            }
        };

Spring线程池ThreadPoolTaskExecutor的优雅装饰

自定义 Spring 任务装饰器 实现 TaskDecorator 接口，封装 TraceID 捕获和恢复逻辑，这是 Spring 官方推荐的线程上下文传递方式。

/**
 * Spring 线程池任务装饰器
 */
public class TraceIdTaskDecorator implements TaskDecorator {

    @Override
    public Runnable decorate(Runnable runnable) {
        // 1. 捕获当前线程的 TraceID 上下文（提交任务时的主线程）
        Map<String, String> context = TraceIdContext.captureContext();

        // 2. 包装任务：异步执行时恢复上下文，执行后清除
        return () -> {
            try {
                TraceIdContext.restoreContext(context); // 恢复 TraceID 到异步线程
                runnable.run(); // 执行原始任务
            } finally {
                TraceIdContext.clear(); // 强制清除，避免 ThreadLocal 泄漏
            }
        };
    }
}

@Configuration
public class ExecutorConfig {
    /**
     * ThreadPoolTaskExecutor 异步线程池
     */
    @Bean("asyncExecutor")
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(8);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(2000);
        executor.setThreadNamePrefix("async-task-");
        // 装饰器
        executor.setTaskDecorator(new TraceIdTaskDecorator());

        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        executor.initialize();
        return executor;
    }
}

封装task

上边的实现，主要用来包装线程池。

我们也可以使用原生线程池，但包装提交的callable或者runnable任务，把业务线程上下文写到异步线程里。

public class ContextThreadUtils {

    public static Runnable wrap(Runnable runnable) {
        // 当前业务线程上下文信息
        Map<String, String> context = TraceIdContext.captureContext();

        // 准备封装原始的Runnable 任务
        return () -> {
            // 异步线程原有的上下文（可能为空）
            Map<String, String> asyncContext = TraceIdContext.captureContext();
            try {
                // 将主线程的上下文设置到当前异步线程
                if (MapUtils.isNotEmpty(context)) {
                    TraceIdContext.restoreContext(context);
                }

                // 执行原始任务
                runnable.run();
            } finally {
                // 恢复原始的异步线程里的上下文
                if (MapUtils.isNotEmpty(asyncContext)) {
                    TraceIdContext.restoreContext(asyncContext);
                }
            }
        };
    }

    public static <T> Callable<T> wrap(Callable<T> callable) {
        // 当前业务线程上下文信息
        Map<String, String> context = TraceIdContext.captureContext();

        return () -> {
            Map<String, String> asyncContext = TraceIdContext.captureContext();
            try {
                // 将主线程的上下文设置到当前异步线程
                if (MapUtils.isNotEmpty(context)) {
                    TraceIdContext.restoreContext(context);
                }
                return callable.call();
            } finally {
                // 恢复原始的异步线程里的上下文
                if (MapUtils.isNotEmpty(asyncContext)) {
                    TraceIdContext.restoreContext(asyncContext);
                }
            }
        };
    }
}

ThreadLocal的问题

ThreadLocal、InheritableThreadLocal 在线程池场景存在缺陷，不天然支持异步上下文透传。

ThreadLocal：仅绑定当前线程，线程池复用线程时，异步任务无法获取父线程上下文，上下文直接丢失。
InheritableThreadLocal：仅在 线程首次创建时 拷贝父线程上下文。线程池线程复用后，后续新提交的任务无法继承最新上下文，且会出现上下文脏数据、任务间数据串扰问题。

当然像上边这种手动处理也可以解决，包装线程池任务时，去做封装。但有多个threadlocal变量时，赋值就麻烦了。这就引出了阿里的TTL组件，解决线程池的透传问题。

TTL 原理

TTL 不依赖线程创建时的一次性继承，而是基于任务级别的快照捕获与恢复，核心是 Capture-Replay-Restore 三段式机制，彻底解决线程复用的数据传递问题：

Capture 捕获（任务提交时）：父线程向线程池提交异步任务的瞬间，主动抓取当前线程的 TTL 上下文快照，绑定到当前任务上，而非绑定线程。
Replay 重放（任务执行前）：线程池工作线程执行任务前，将任务绑定的上下文快照，恢复到当前复用线程的 ThreadLocal 中。
Restore 恢复（任务结束后）：任务执行完毕后，清空/还原线程原有上下文。彻底避免线程复用导致的上下文残留、数据串扰。

优势：上下文跟随任务，而非跟随线程，适配所有池化复用线程场景。

TTL使用方式

一般分两种：

包装线程池
包装任务

public class TtlDemo {
    // 定义TTL上下文（替代原生ThreadLocal）
    private static final TransmittableThreadLocal<String> TRACE_ID = new TransmittableThreadLocal<>();

    public static void main(String[] args) throws InterruptedException {
        // 1. 创建线程池，并使用TTL包装
        ExecutorService pool = TtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(1));

        // 2. 父线程设置上下文
        TRACE_ID.set("test-001");
        System.out.println("主线程上下文：" + TRACE_ID.get());

        // 3. 第一次提交异步任务，复用线程池线程
        pool.execute(() -> System.out.println("异步任务1获取上下文：" + TRACE_ID.get()));

        // 覆盖父线程上下文，模拟不同请求的上下文更新
        TRACE_ID.set("test-002");

        // 4. 第二次提交任务，验证线程复用后的最新上下文透传
        pool.execute(() -> System.out.println("异步任务2获取上下文：" + TRACE_ID.get()));

        pool.shutdown();
    }

    public static void main(String[] args) throws InterruptedException {
        // 原生线程池，未做TTL包装
        ExecutorService pool = Executors.newFixedThreadPool(1);

        // 父线程设置上下文
        TRACE_ID.set("test-001");

        // 手动包装任务，TTL即可正常生效
        pool.execute(TtlRunnable.get(() -> {
            System.out.println("手动包装任务获取上下文：" + TRACE_ID.get());
        }));
    }

}

java