异步线程上下文传递机制

概述

本文档详细描述了框架中 @Async 异步线程场景下 ThreadContextHolder 上下文（用户信息、租户信息、链路ID）的自动传递机制、实现原理及使用方法。

问题背景

在传统的 ThreadLocal 使用中，存在以下问题：

1. 普通 ThreadLocal：值绑定在当前线程对象上，线程池中的工作线程无法获取主线程设置的值
2. InheritableThreadLocal：仅在创建子线程时能继承父线程值，但线程池中的线程是复用的，不是任务提交时创建，因此在线程池场景下失效
3. 线程复用导致数据污染：任务执行完后 ThreadLocal 值未清理，会影响线程池下一个任务

框架通过 TransmittableThreadLocal (TTL) + TaskDecorator 方案解决了上述问题。

架构设计

相关模块与文件

核心组件关系

┌─────────────────────────────────────────────────────────────────┐
│                      Web 请求主线程                              │
│  ┌──────────────────┐    ┌──────────────────────────────────┐  │
│  │ ContextHolderFilter │──▶│ ThreadContextHolder.setUserInfo() │  │
│  └──────────────────┘    └──────────────┬───────────────────┘  │
│                                         │                       │
│                          ┌──────────────▼───────────────────┐  │
│                          │  调用 @Async 方法 / 提交线程池任务  │  │
│                          └──────────────┬───────────────────┘  │
└─────────────────────────────────────────┼───────────────────────┘
                                          │
                                          ▼
                            ┌──────────────────────────┐
                            │ DefaultTaskDecorator     │
                            │  ┌─────────────────────┐ │
                            │  │ TtlRunnable.get()   │ │
                            │  │ 【捕获】TTL 快照    │ │
                            │  └─────────────────────┘ │
                            └─────────────┬────────────┘
                                          │
                                          ▼
┌─────────────────────────────────────────┼───────────────────────┐
│                    线程池工作线程         │                       │
│                          ┌──────────────▼───────────────────┐  │
│                          │ TtlRunnable.run()                │  │
│                          │  ┌─────────────────────────────┐ │  │
│                          │  │ 1. 备份子线程原有 TTL 值     │ │  │
│                          │  │ 2. 【回放】主线程 TTL 快照   │ │  │
│                          │  │ 3. 执行业务逻辑              │ │  │
│                          │  │    → getUserInfo() 有值      │ │  │
│                          │  │ 4. 恢复子线程原有 TTL 值     │ │  │
│                          │  └─────────────────────────────┘ │  │
│                          └──────────────┬───────────────────┘  │
│                                         │                       │
│                          ┌──────────────▼───────────────────┐  │
│                          │ finally 清理资源                  │  │
│                          │  - MDC.remove(traceId)           │  │
│                          │  - PageHelper.clearPage()        │  │
│                          └──────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

实现原理详解

1. ThreadContextHolder：定义可传递的上下文

ThreadContextHolder.java 中对不同数据采用不同的 ThreadLocal 策略：

@UtilityClass
public class ThreadContextHolder {
    /**
     * 用户信息 - 使用 TransmittableThreadLocal
     * ✅ 支持父子线程/线程池场景下的数据传递
     * 包含：userId、renterId、tenantId 等租户隔离关键字段
     */
    private final ThreadLocal<UserInfo> USER_INFO_THREAD_LOCAL = new TransmittableThreadLocal<>();

    /**
     * 分布式事务信息 - 使用普通 ThreadLocal
     * ❌ 不支持跨线程传递（设计如此，避免事务跨线程问题）
     */
    private final ThreadLocal<TransactionInfo> TRANSACTION_THREAD_LOCAL = new ThreadLocal<>();

    /**
     * 业务信息 - 使用普通 ThreadLocal
     * ❌ 不支持跨线程传递
     */
    private static final ThreadLocal<BusinessInfo> BUSINESS_INFO_THREAD_LOCAL = ThreadLocal.withInitial(BusinessInfo::new);
}

2. DefaultTaskDecorator：任务装饰核心逻辑

DefaultTaskDecorator.java 实现了 Spring 的 TaskDecorator 接口：

@Slf4j
public class DefaultTaskDecorator implements TaskDecorator {

    /**
     * 线程池执行任务时，将当前线程的 TransmittableThreadLocal 传递给子线程
     * 注意：没有传递事务，注意以后这个方法也不能传递事务，
     *       需要传递事务的情况要另外实现
     *
     * @param runnable 原始任务
     * @return 包装后的代理任务
     */
    @Override
    public Runnable decorate(Runnable runnable) {
        // 1. 捕获主线程的 traceId（用于链路追踪）
        String traceId = MDC.get(CommonConstants.TRACE_ID);
  
        // 2. 核心：使用 TtlRunnable 包装，捕获当前线程 TTL 快照
        TtlRunnable ttlRunnable = TtlRunnable.get(runnable);
  
        // 3. 返回包装后的任务，在线程池线程中执行
        return () -> {
            try {
                // 4. 设置/传递 traceId
                MDC.put(CommonConstants.TRACE_ID, 
                    Optional.ofNullable(traceId).orElse(IdUtil.getSnowflake().nextIdStr()));
        
                // 5. TtlRunnable.run() 内部自动完成：
                //    - 备份子线程原有 TTL 值
                //    - 回放主线程 TTL 快照到子线程
                //    - 执行业务 runnable
                //    - 恢复子线程原有 TTL 值
                ttlRunnable.run();
            } finally {
                // 6. 清理资源，避免线程复用污染
                MDC.remove(CommonConstants.TRACE_ID);
                PageHelper.clearPage();
            }
        };
    }
}

3. TaskExecutorConfigurations：Spring @Async 线程池配置

TaskExecutorConfigurations.java 根据 JVM 线程模型条件化创建默认异步执行器，并统一设置 TaskDecorator：

@Configuration(proxyBeanMethods = false)
public class TaskExecutorConfigurations {

    /**
     * 虚拟线程模式（Java 21+，启用 spring.threads.virtual.enabled=true）
     * 使用 SimpleAsyncTaskExecutor，每个任务一个虚拟线程
     */
    @Bean(CommonConstants.DEFAULT_ASYNC_EXECUTOR_BEAN_NAME)
    @Primary
    @ConditionalOnThreading(Threading.VIRTUAL)
    SimpleAsyncTaskExecutor applicationTaskExecutorVirtualThreads(SimpleAsyncTaskExecutorBuilder builder) {
        SimpleAsyncTaskExecutor executor = builder.build();
        ThreadPoolUtils.init(executor, new DefaultTaskDecorator());
        return executor;
    }

    /**
     * 平台线程模式（默认）
     * 使用 ThreadPoolTaskExecutor，传统线程池复用模式
     */
    @Bean(CommonConstants.DEFAULT_ASYNC_EXECUTOR_BEAN_NAME)
    @Primary
    @ConditionalOnThreading(Threading.PLATFORM)
    ThreadPoolTaskExecutor applicationTaskExecutor(ThreadPoolTaskExecutorBuilder threadPoolTaskExecutorBuilder) {
        ThreadPoolTaskExecutor executor = threadPoolTaskExecutorBuilder.build();
        ThreadPoolUtils.init(executor, new DefaultTaskDecorator());
        return executor;
    }

    /**
     * Spring Boot 自动配置的线程池定制器 - 平台线程
     */
    @Bean
    public ThreadPoolTaskExecutorCustomizer customizeThreadPool() {
        return taskExecutor -> {
            taskExecutor.setTaskDecorator(new DefaultTaskDecorator());
            taskExecutor.setThreadNamePrefix("imom-task-p-");
        };
    }

    /**
     * Spring Boot 自动配置的线程池定制器 - 虚拟线程
     */
    @Bean
    public SimpleAsyncTaskExecutorCustomizer customizeSimpleThreadPool() {
        return taskExecutor -> {
            taskExecutor.setTaskDecorator(new DefaultTaskDecorator());
            taskExecutor.setThreadNamePrefix("imom-task-v-");
        };
    }
}

传递内容对照表

使用方法

1. 基础使用：直接使用 @Async

框架已经自动配置好了默认的异步线程池，业务代码直接使用 Spring 的 @Async 注解即可，无需任何额外配置：

import com.sie.mbm.mom.framework.common.ttl.ThreadContextHolder;
import com.sie.mbm.mom.framework.common.ttl.UserInfo;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class MyBusinessService {

    /**
     * 异步方法 - 自动传递 ThreadContextHolder 上下文
     */
    @Async
    public void asyncMethod() {
        // ✅ 在异步线程中可以正常获取用户信息
        UserInfo user = ThreadContextHolder.getUserInfo();
        if (user != null) {
            String userId = user.getId();
            String renterId = user.getRenterId();
            String tenantId = user.getTenantId();
            log.info("异步线程获取到用户: userId={}, renterId={}, tenantId={}", 
                     userId, renterId, tenantId);
    
            // ✅ 此处可以正常使用需要租户隔离的 Mapper/Service
            // （因为 DbFilterUtil 等组件也是从 ThreadContextHolder 取租户信息）
        }
    }
}

2. 使用 ThreadPoolUtils 提交任务

如果需要通过工具类手动提交任务到默认线程池：

import com.sie.mbm.mom.framework.common.util.ThreadPoolUtils;
import com.sie.mbm.mom.framework.common.ttl.ThreadContextHolder;

public class SomeService {

    public void submitTask() {
        // 方式1：execute 提交 Runnable
        ThreadPoolUtils.execute(() -> {
            // ✅ 上下文已传递
            var user = ThreadContextHolder.getUserInfo();
            // 业务逻辑
        });

        // 方式2：submit 提交 Runnable，返回 Future
        var future = ThreadPoolUtils.submit(() -> {
            // 业务逻辑
        });

        // 方式3：submitCompletable 提交 CompletableFuture
        var completableFuture = ThreadPoolUtils.submitCompletable(() -> {
            // 业务逻辑
            return "result";
        });
    }
}

3. 自定义线程池时传递上下文

如果业务需要自定义线程池（非 @Async 默认线程池），需要手动设置 DefaultTaskDecorator：

import com.sie.mbm.mom.framework.core.concurrent.DefaultTaskDecorator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
public class CustomThreadPoolConfig {

    @Bean("myCustomExecutor")
    public ThreadPoolTaskExecutor myCustomExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("my-custom-p-");
  
        // ⚠️ 关键：设置 DefaultTaskDecorator 保证上下文传递
        executor.setTaskDecorator(new DefaultTaskDecorator());
  
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

使用时指定线程池名称：

@Async("myCustomExecutor")
public void asyncMethodWithCustomExecutor() {
    // ✅ 上下文也能正常传递
    var user = ThreadContextHolder.getUserInfo();
}

4. Callable 场景的上下文传递

如果需要使用 Callable 带返回值的任务，框架的 DefaultTaskDecorator 只处理了 Runnable。对于 Callable，可以使用 TTL 提供的 TtlCallable：

import com.alibaba.ttl.TtlCallable;
import java.util.concurrent.Callable;

public <T> T submitWithTtl(Callable<T> task) throws Exception {
    // 包装 Callable，类似 TtlRunnable
    Callable<T> ttlCallable = TtlCallable.get(task);
    return executor.submit(ttlCallable).get();
}

完整请求流程示例

1. HTTP 请求进入
   ↓
2. ContextHolderFilter.doFilter()
   ├─ 从请求头/Token 解析用户信息
   └─ ThreadContextHolder.setUserInfo(userInfo)  ✅ 设置用户信息到 TTL
   ↓
3. Controller 调用 Service
   ↓
4. Service 调用 @Async 方法 / ThreadPoolUtils.submit()
   ↓
5. DefaultTaskDecorator.decorate(runnable)  (在主线程执行)
   ├─ 捕获 MDC traceId
   └─ TtlRunnable.get(runnable)  ✅ 捕获主线程 TTL 快照
   ↓
6. 任务提交到线程池队列
   ↓
7. 线程池工作线程取出任务执行
   ↓
8. TtlRunnable.run()  (在子线程执行)
   ├─ 备份子线程原有 TTL 值
   ├─ 【回放】主线程 TTL 值到子线程  ✅ 此时子线程 getUserInfo() 有值
   ├─ 执行用户业务逻辑：
   │  ├─ ThreadContextHolder.getUserInfo() → 正常获取
   │  ├─ DbFilterUtil 自动租户隔离 → 正常工作
   │  └─ 缓存操作自动携带 renterId/tenantId → 正常工作
   └─ 恢复子线程原有 TTL 值
   ↓
9. DefaultTaskDecorator finally 块
   ├─ MDC.remove(traceId)
   └─ PageHelper.clearPage()
   ↓
10. 请求返回，ContextHolderFilter finally 块
    └─ ThreadContextHolder.clearAll()  ✅ 清理主线程上下文

注意事项

1. 事务不能跨线程传递

• 设计上 TRANSACTION_THREAD_LOCAL 使用普通 ThreadLocal，不跨线程传递
• 异步方法中的数据库操作不在原事务中，需要独立事务处理
• 如果需要异步执行数据库操作，建议：
  • 在异步方法内部使用 @Transactional(propagation = Propagation.REQUIRES_NEW)
  • 或者先在主线程完成事务操作，再异步处理非事务逻辑

2. 自定义线程池必须设置 TaskDecorator

• 业务代码中自己 new 出来的 ThreadPoolExecutor 默认不会自动传递上下文
• 必须调用 setTaskDecorator(new DefaultTaskDecorator()) 才能传递
• 推荐使用框架已配置的 ThreadPoolUtils 提交任务，无需额外配置

3. 异步方法必须通过 Spring 代理调用

• @Async 注解的方法必须通过 Spring 注入的代理对象调用才生效
• 同类内部直接调用（this.asyncMethod()）不会走代理，异步不生效，上下文传递也无从谈起

// ❌ 错误：同类内部调用，@Async 不生效
@Service
public class OrderService {
    public void createOrder() {
        this.asyncProcess();  // 直接调用，不是代理调用
    }
  
    @Async
    public void asyncProcess() { ... }
}

// ✅ 正确：通过注入的自身代理调用，或放到单独的 Bean
@Service
public class OrderService {
    @Autowired
    @Lazy
    private OrderService self;  // 注入自身代理
  
    public void createOrder() {
        self.asyncProcess();  // 通过代理调用，@Async 生效
    }
  
    @Async
    public void asyncProcess() { ... }
}

4. TtlRunnable 不是银弹

• 只有使用 TransmittableThreadLocal 存储的值才会被传递
• 普通 ThreadLocal 的值不会被自动传递，如需传递需要在 TaskDecorator 中手动处理

5. 线程池上下文清理

• 框架的 DefaultTaskDecorator 已经在 finally 块中清理了 MDC 和 PageHelper
• 如果业务代码自定义了其他 ThreadLocal，需要自行在任务结束后清理，避免线程复用污染

常见问题

Q: 异步方法能拿到租户信息（renterId/tenantId）吗？

能。UserInfo 中包含 renterId 和 tenantId，通过 TTL 自动传递，DbFilterUtil 等租户隔离组件在异步线程中可以正常工作。

Q: 异步方法中使用 @Transactional 会怎么样？

• 异步方法运行在独立线程，和主线程的事务是两个独立的事务
• 默认传播级别是 REQUIRED，会在异步线程开启新事务（因为异步线程没有已有事务）
• 建议异步方法需要事务时显式使用 @Transactional(propagation = Propagation.REQUIRES_NEW)

API 参考

ThreadContextHolder 常用方法

ThreadPoolUtils 常用方法