Java中如何确保CompletableFuture正确传播ThreadLocal上下文

ThreadLocal的核心特性是线程绑定，它的上下文只存在于当前线程的内存空间中。你用CompletableFuture提交任务到线程池时，任务会在线程池的复用线程中执行，这些线程和处理请求的原线程完全是不同的实例，原线程的ThreadLocal上下文不会自动复制到异步线程里，所以异步任务中DatabaseContextHolder的ctx会为空。

你之前尝试的synchronized和更换执行器都解决不了这个问题：synchronized是保证多线程对共享资源的互斥访问，和ThreadLocal的上下文传递无关；不管用哪种线程池，只要是跨线程执行任务，ThreadLocal都不会自动传递。

方案1：手动捕获并传递上下文（快速实现） #

在创建异步任务前，先捕获当前线程的数据源上下文，然后在异步任务执行时手动设置，执行完成后必须清理（线程池线程会复用，不清理会导致上下文污染后续任务）。

修改你的groupIds.forEach代码块：

groupIds.forEach(id -> {
    // 捕获当前请求线程的数据源上下文
    Optional<DataSourceType> currentDataSource = DatabaseContextHolder.peekDataSource();
    
    CompletableFuture<List<ReportGroupScheduleExecutionBean>> future = CompletableFuture.supplyAsync(
            () -> {
                try {
                    // 将上下文设置到当前异步线程
                    currentDataSource.ifPresent(DatabaseContextHolder::setCtx);
                    return reportService.getRouteScheduleExecutionReportByGroup(
                            new ReportInfoBean(dateFrom, dateTo, account, userBean.getUserTimeZone(), zoneTimeFormatter), 
                            id);
                } catch (NoDataException e) {
                    throw new RuntimeException(e);
                } finally {
                    // 执行完毕后恢复上下文，避免线程池线程复用导致的上下文泄漏
                    if (currentDataSource.isPresent()) {
                        DatabaseContextHolder.restoreCtx();
                    }
                }
            }, executorService);
    futures.add(future);
});

方案2：包装线程池，自动传递上下文（优雅复用） #

如果你的代码中有很多异步任务需要传递上下文，手动处理会很繁琐。可以写一个线程池装饰器，自动捕获提交任务时的ThreadLocal上下文，在任务执行时注入，执行后清理。

1. 实现上下文感知的ExecutorService

public class ContextAwareExecutorService implements ExecutorService {
    private final ExecutorService delegate;

    public ContextAwareExecutorService(ExecutorService delegate) {
        this.delegate = delegate;
    }

    // 包装Runnable任务，自动传递上下文
    private Runnable wrapRunnable(Runnable task) {
        Optional<DataSourceType> currentCtx = DatabaseContextHolder.peekDataSource();
        return () -> {
            try {
                currentCtx.ifPresent(DatabaseContextHolder::setCtx);
                task.run();
            } finally {
                currentCtx.ifPresent(ds -> DatabaseContextHolder.restoreCtx());
            }
        };
    }

    // 包装Callable任务，自动传递上下文
    private <T> Callable<T> wrapCallable(Callable<T> task) {
        Optional<DataSourceType> currentCtx = DatabaseContextHolder.peekDataSource();
        return () -> {
            try {
                currentCtx.ifPresent(DatabaseContextHolder::setCtx);
                return task.call();
            } finally {
                currentCtx.ifPresent(ds -> DatabaseContextHolder.restoreCtx());
            }
        };
    }

    // 重写ExecutorService的所有方法，将任务包装后提交给原线程池
    @Override
    public void shutdown() {
        delegate.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        return delegate.shutdownNow();
    }

    @Override
    public boolean isShutdown() {
        return delegate.isShutdown();
    }

    @Override
    public boolean isTerminated() {
        return delegate.isTerminated();
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return delegate.awaitTermination(timeout, unit);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return delegate.submit(wrapCallable(task));
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return delegate.submit(wrapRunnable(task), result);
    }

    @Override
    public Future<?> submit(Runnable task) {
        return delegate.submit(wrapRunnable(task));
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        return delegate.invokeAll(tasks.stream().map(this::wrapCallable).collect(Collectors.toList()));
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        return delegate.invokeAll(tasks.stream().map(this::wrapCallable).collect(Collectors.toList()), timeout, unit);
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        return delegate.invokeAny(tasks.stream().map(this::wrapCallable).collect(Collectors.toList()));
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return delegate.invokeAny(tasks.stream().map(this::wrapCallable).collect(Collectors.toList()), timeout, unit);
    }

    @Override
    public void execute(Runnable command) {
        delegate.execute(wrapRunnable(command));
    }
}

2. 替换原线程池实例

把原来的线程池替换成包装后的实例：

private final ExecutorService executorService = new ContextAwareExecutorService(Executors.newCachedThreadPool());

这样所有提交到这个线程池的任务都会自动传递ThreadLocal上下文，无需手动处理，同时保证了线程池线程的上下文干净，不会污染后续任务。

额外注意事项 #

• 线程池线程复用导致的上下文泄漏是必须要避免的，所以一定要在finally块中清理上下文。
• 如果是Spring环境，还可以使用ThreadPoolTaskExecutor的setTaskDecorator方法来实现类似的上下文传递，无需自己写装饰器。

内容的提问来源于stack exchange，提问作者VegetaSan