多线程与线程在项目中的使用笔记
|
20
|
0
|
new-mall

2155 字
|
12 分钟

多线程与线程在项目中的使用笔记

先说结论：业务代码里没有手写 new Thread、没有 @Async、没有自定义线程池、没有 CompletableFuture/parallelStream。项目的”多线程”主要是框架托管的线程模型 + 两处 ThreadLocal。本文按真实代码梳理：ThreadLocal、Tomcat 请求线程池、@Scheduled 定时线程、RabbitMQ 消费者线程、WebFlux 事件循环线程，最后给一份生产级 @Async 线程池方案。

一、两种并发模型并存

服务 模型 线程特征
大部分服务（admin/product/order/cart…） Spring MVC（Servlet 阻塞） Tomcat 线程池，一个请求占一个工作线程，线程内同步执行
mall-user 的 AI 客服 Spring WebFlux（响应式） 少量 Netty 事件循环线程，SSE + Flux 异步推送
mall-gateway Spring Cloud Gateway（WebFlux） 同样跑在 Netty event-loop 上

二、ThreadLocal —— 项目最核心的”线程”用法

它不创建线程，而是给每个线程一份独立变量副本，用来在同一次请求的多个方法/切面间传数据，又互不干扰。

位置 存什么 作用
LoginContext（ThreadLocal<Map>） 当前登录用户 id/name 业务代码随时取当前用户，不用层层传参
OperLogAspect（ThreadLocal<OperLog>） 本次请求的操作日志对象 把 @Before 建的对象传到 @AfterReturning/@AfterThrowing

为什么必须线程隔离：Tomcat 线程池里同一个线程会被多个请求复用，不用 ThreadLocal 会串数据。

用完必须 remove（高频考点）

LoginInterceptor 在 preHandle 里 set，在 afterCompletion 里 removeLoginInfo()；日志切面在结束通知里也 remove。不清理的两个后果：

  1. 线程复用污染：上一个请求的用户 id 残留，下一个请求读到别人身份（越权）；
  2. 内存泄漏：对象一直被线程引用无法回收。

ThreadLocal 不跨线程（本项目多处受此约束）

请求线程里放的用户信息，换个线程就拿不到：

  • MQ 消费者、@Scheduled 定时任务、@Async 线程池里 LoginContext.getLoginInfo() 是 null；
  • 所以秒杀消息体 SeckillMessage 要显式带 userId，不能让消费者去读 ThreadLocal；
  • Feign 拦截器 DefaultFeignConfig 也先判 null——定时任务发起的 Feign 调用没有用户上下文，直接跳过透传。

三、Tomcat 请求线程池（MVC）

每个进来的 HTTP 请求由 Tomcat 从工作线程池取一个线程执行 Controller → Service → Mapper，线程内是同步阻塞调用（查库、Feign 等待时线程挂起）。

含义：

  • LoginContext 能在一次请求内任意方法取到用户，正是因为全程在同一个 Tomcat 线程；
  • 高并发下若大量线程都阻塞在慢查询/慢 Feign 上，线程池打满会拒绝新请求——这是要用缓存（Redis）、异步削峰（MQ）的根本原因。

四、@Scheduled 定时任务线程

mall-service/.../job/DeleteImageJob.java：

@Scheduled(cron = "${oss.cleanup-cron:0 0/2 * * * *}")  // 每 2 分钟清理 OSS 孤儿图片（Spring 6 段 cron，无 ?）
public void deleteImage() { ... }

@EnableScheduling 标在 ServiceApplication。它默认跑在 Spring 的单线程 TaskScheduler 上：

  • 多个 @Scheduled 方法默认串行，一个卡死会拖慢其他所有定时任务；
  • 它是独立调度线程，没有 HTTP 请求上下文（所以里面 Feign 调用要在白名单/内部端点放行，也读不到登录用户）；
  • 生产若任务多，应自定义 ThreadPoolTaskScheduler 提供线程池。

五、RabbitMQ 消费者线程（异步削峰的真正落点）

两个监听器：OrderStatusListener（订单超时延时消息）、SeckillOrderListener（秒杀下单）：

@RabbitListener(bindings = @QueueBinding(...))
public void onSeckillOrder(SeckillMessage msg) { ... }

消息处理不在 Tomcat 请求线程，而在 SimpleMessageListenerContainer 维护的消费者线程池里。意义：

  • 秒杀抢购接口只做 Redis 预减 + 发 MQ 就立刻返回，真正的查库、扣库存、建订单在消费者线程按数据库能力平稳消费 → 削峰；
  • 这是典型的”请求线程与业务处理线程解耦”；
  • 消费者是独立线程，读不到请求的 ThreadLocal，靠消息体传 userId；
  • 并发度可用 concurrency = "5-10" 调（项目当前用默认单消费者、AUTO ack，生产可调）。

六、WebFlux 事件循环线程（AI 客服 / 网关）

ChatController 返回 Flux<ServerSentEvent>，DeepSeekService 用 WebClient 调 DeepSeek：

webClient.post().retrieve().bodyToFlux(String.class);   // 少量 Netty event-loop 线程

特点：少量线程处理大量并发连接，靠非阻塞 + 回调/数据流而不是”一请求一线程”。

铁律：绝不能在事件循环线程上做阻塞操作（JDBC、Thread.sleep、.block），否则一个阻塞拖垮所有响应式请求。项目的正确处理在 ChatService：

// 把阻塞的 DB 调用包进 fromCallable，调度到 boundedElastic 弹性线程池，不卡事件循环
Mono<Flux<ServerSentEvent<String>>> completion = Mono.fromCallable(() -> {
    ... saveConversation(userId, sessionId, messages);   // 阻塞的 JDBC
});

非流式场景则用 .block()（DeepSeekService）把响应式结果桥接回普通同步代码。网关的 AuthGlobalFilter 也是返回 Mono<Void> 的响应式过滤器。

七、框架内部线程（业务无感，了解即可）

  • Redisson：底层 Netty event-loop 发送 Redis 命令；看门狗用后台定时线程给锁自动续期；
  • HikariCP：数据库连接池，连接获取/归还的调度；
  • Lettuce / Netty：Redis 客户端的 IO 线程；
  • Spring AOP 代理本身不开线程，但让”事务、锁、缓存”在同一线程上按切面顺序执行。

八、生产级 @Async + 自定义线程池方案

秒杀下单不要从 MQ 改成 @Async（会丢掉消息持久化、削峰、重试、跨机消费能力，宕机即丢单，详见第九节）。@Async 适合可丢失、短平快的进程内异步，本项目最佳落点是把 OperLogAspect 里同步落库的操作日志改成异步。

8.1 开启异步

@EnableAsync
@Configuration
public class AsyncEnableConfig {}

8.2 自定义线程池（IO 密集型）

@Configuration
public class ThreadPoolConfig {
    private static final int CPU = Runtime.getRuntime().availableProcessors();

    @Bean("bizAsyncExecutor")
    public ThreadPoolTaskExecutor bizAsyncExecutor() {
        ThreadPoolTaskExecutor e = new ThreadPoolTaskExecutor();
        e.setCorePoolSize(2 * CPU);       // IO 密集约 2N；CPU 密集才 N+1
        e.setMaxPoolSize(4 * CPU);
        e.setQueueCapacity(1000);         // 必须有界，防止任务堆积 OOM
        e.setKeepAliveSeconds(60);
        e.setThreadNamePrefix("biz-async-");  // 排障能认出线程
        e.setAllowCoreThreadTimeOut(true);
        e.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 反压
        e.setWaitForTasksToCompleteOnShutdown(true);  // 优雅停机，等任务做完
        e.setAwaitTerminationSeconds(30);
        e.setTaskDecorator(new ContextCopyingDecorator()); // 透传 ThreadLocal/MDC
        e.initialize();
        return e;
    }
}

8.3 ThreadLocal/MDC 跨线程透传装饰器（最容易漏）

public class ContextCopyingDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        // 提交线程（HTTP 线程）此刻抓快照
        Map<String, Object> login = LoginContext.getLoginInfo();
        Map<String, Object> loginSnapshot = login == null ? null : Map.copyOf(login);
        Map<String, String> mdcSnapshot = MDC.getCopyOfContextMap();
        return () -> {                      // 线程池线程执行
            try {
                if (loginSnapshot != null) LoginContext.setLoginInfo(new HashMap<>(loginSnapshot));
                if (mdcSnapshot != null) MDC.setContextMap(mdcSnapshot);
                runnable.run();
            } finally {
                LoginContext.removeLoginInfo();  // 线程复用，必须清理
                MDC.clear();
            }
        };
    }
}

8.4 使用与两个坑

@Async("bizAsyncExecutor")          // 指定线程池
public void saveOperLogAsync(OperLog log) { operLogMapper.insert(log); }
  1. 自调用失效：必须跨 Bean 调用，同类 this.saveOperLogAsync() 不走代理，@Async 不生效（和 @Transactional 同理）；
  2. void 方法异常被吞：要配异常处理器：
@Configuration
public class AsyncExceptionConfig implements AsyncConfigurer {
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) -> log.error("异步任务失败 method={}", method.getName(), ex);
    }
}

8.5 参数口径

  • 扩容顺序：核心线程满 → 进队列 → 队列满才扩到 max → 还满才触发拒绝策略；所以有界队列容量决定 max 何时生效；
  • 拒绝策略：CallerRunsPolicy（调用线程自己执行=反压，保不丢但拖慢主流程）/ AbortPolicy（快速失败）/ 自定义落库补偿；绝不用无界队列（峰值 OOM）；
  • waitForTasksToCompleteOnShutdown=true 防止发布重启时队列任务被直接砍断。

九、为什么秒杀用 MQ 而不是 @Async

维度 RabbitMQ（项目现状） @Async 线程池
宕机 消息持久化，重启继续消费 任务在内存，崩溃即丢单
削峰/背压 峰值堆 Broker，按 concurrency 平稳消费 有界队列满只能拒绝或反压
失败重试 nack/requeue/死信 抛异常任务即丢，需自建补偿
水平扩展 多实例共消费同一队列 每实例各自内存池，无法跨机分摊

下单涉及钱和库存、绝不能丢，必须 MQ；@Async 只用于操作日志这类丢了无伤大雅的辅助动作。

十、面试一句话总结

项目没手写多线程，用的是框架托管线程：Tomcat 请求线程池配 ThreadLocal 传登录态和操作日志（用完 remove 防线程复用污染）；@Scheduled 单线程跑 OSS 清理；RabbitMQ 消费者线程池异步处理超时关单和秒杀下单做削峰；AI 客服和网关用 Netty 事件循环线程，阻塞 DB 调用用 Mono.fromCallable 挪到弹性线程池。关键认知是 ThreadLocal 不跨线程，所以 MQ/异步要显式传 userId；核心交易异步用 MQ，@Async 只用于日志等可丢失场景，并要配线程池参数、拒绝策略、上下文透传和优雅停机。

上一篇
下一篇