企业官网建设流程全解析

说个真实场景吧：你开了一家餐厅，生意火爆，但坚持让客人站着等饭做好才落座——菜还没上，座位已经被站着等的人全占完了。这是我接手一个Java AI应用时脑子里蹦出来的第一幅画面。项目不复杂，就是给业务方做一个基于大模型的智能问答接口，但线上表现很割裂：Tomcat线程池动不动被打满，AI接口平均耗时2秒，QPS只要一过30就开始大量超时，日志里全是“Connection pool exhausted”和“Task rejected”。

这篇文章就围绕Java AI应用的异步化与高并发设计来讲。你未必在写大模型应用，只要是Java后端、要调用慢外部依赖、想扛住高并发，思路都通用。内容包括四块：为什么AI场景必须异步化，Java生态里几种异步方案怎么选，高并发设计的几个关键环节，以及一个能直接参考的实操案例。结尾我会把踩过的坑列一份速查表，帮你有针对性地排查问题。

1. 为什么AI应用比传统Web应用更需要异步化

1.1 AI应用的延迟特征：从毫秒级到秒级甚至分钟级

传统业务接口讲究低延迟，100毫秒以内是常态，超过1秒用户就开始焦虑。AI应用完全不同。一次模型推理动辄几百毫秒到几秒，如果涉及RAG（检索增强生成），要先做向量检索、再拼提示词、再调模型，走完一轮轻松5到10秒。要是做成Agent形态，大模型要多次调用工具、来回规划，耗时到几十秒也正常。

这里有一个核心矛盾：HTTP连接和线程是稀缺资源，而AI服务天然“慢”。同步处理模式里，一个请求占住一个Tomcat线程不放手，线程池只有200个，意味着一瞬间只能同时处理200个请求。AI接口慢，线程就大规模阻塞。用户侧看到的是请求排队、超时、5xx，而服务端CPU使用率可能并不高——因为线程都在等，根本没在算。

我做过一个小实验，把一个内部AI接口的耗时从800ms调到8秒，其他条件不变，系统能承受的QPS直接降了大概9倍。这就是为什么AI应用必须把“等待”和“计算”解耦，不能让链路里最慢的那一环决定整个吞吐量。

1.2 同步阻塞的本质：线程不是等不起，是太贵

要理解异步化的价值，先得算一笔账。JVM里每个线程默认栈大小1MB，注意这是虚拟内存，但它承载的并发能力确实有限。一个4核8G的机器，JVM堆设置4G，线程数开到500到800通常就已经偏大，继续加线程会导致上下文切换开销暴涨，GC也跟着恶化。

而且线程不是免费工人，它是“有编制的员工”——创建、销毁、调度都有成本。Tomcat通过线程池复用线程，本质上就是为了摊薄这些成本。但复用的前提是线程能很快空出来接下一个任务。AI应用把线程按在那里等外部HTTP响应，相当于员工站在快递柜前等包裹，一等等半天，活全积压了。

异步化的目标，恰恰是让线程只做“发起调用”和“处理结果”两件短活，把漫长的等待时间交给底层的非阻塞IO和回调机制，让线程去服务其他请求。

1.3 异步化要解决什么：线程利用率、响应速度和削峰

具体来说，异步化在AI场景下能带来三方面收益。

第一是线程利用率。同样200个线程，同步模式只能同时处理200个在途请求，异步模式下可以同时挂着几千个在途请求，因为线程在等待结果时已经被释放了。JVM本身不感知HTTP连接是否活跃，但它感知线程，所以提高线程复用，就是提高并发承载量。

第二是响应速度。异步模型允许一个请求拆分多个子任务并行执行。比如Agent场景需要同时调两个工具，同步写就是串行等两次，异步可以并行等一次，耗时从T1加T2变成max(T1, T2)。这对AI链路总延迟的改善立竿见影。

第三是削峰填谷。异步化常常配合消息队列一起用。用户点击提问后，请求先返回“已受理”，后台异步任务慢慢执行，查询结果通过轮询或SSE推送返回。这样业务高峰期的突发流量不会直接压垮模型服务，而是进入队列排队，系统吞吐量反而更稳。

2. Java生态异步方案横向对比：CompletableFuture、虚拟线程、消息队列到底怎么选

2.1 CompletableFuture：JDK原生的异步组合方案

CompletableFuture是Java 8引入的，它弥补了Future的短板——不能手动编排任务、不能回调、不能组合。

我在AI应用里最常用的几个方法：

• supplyAsync()：提交一个异步任务，返回CompletableFuture
• thenApplyAsync()：上一个任务完成后继续处理，支持异步执行
• allOf()：等待多个异步任务全部完成，常用于并行调用多个模型
• exceptionally()：捕获异常并返回一个默认值，做降级

示例代码如下：

public CompletableFuture<AnswerResult> parallelAnswer(String question) { CompletableFuture<RetrievalResult> retrievalFuture = CompletableFuture.supplyAsync(() -> retrievalService.search(question), searchExecutor); CompletableFuture<LlmResult> llmFuture = CompletableFuture.supplyAsync(() -> llmService.generate(question), llmExecutor); return retrievalFuture .thenCombine(llmFuture, (retrieval, llm) -> new AnswerResult(retrieval.getContexts(), llm.getAnswer())) .exceptionally(ex -> { log.error("parallel answer failed", ex); return AnswerResult.fallback(); }); }

这段代码的作用是：向量检索和大模型生成这两个独立步骤可以并行执行，而不是检索完再生成。对RAG场景来说，时间从两个步骤相加缩短为两者中的较大值。

使用CompletableFuture有两点必须注意。默认使用的ForkJoinPool.commonPool非常坑，它的并行度是CPU核数减1，AI场景下大批量任务会把它挤爆，所以务必自建线程池传入。再有就是异常处理不能遗漏，异步链路里的异常不会像同步代码那样自动向上抛，要在exceptionally()或者handle()里显式兜底，否则用户请求会“永久悬挂”。

2.2 Spring的@Async和线程池配置：异步落地的第一步

如果你的项目是Spring Boot，最省事的异步入口就是@Async。但这里有个高频误区：随便贴一个@Async注解就完事了，结果方法没异步执行。原因是Spring的异步基于AOP代理，而代理对this内部调用不生效，只有通过Spring容器调用bean方法时才会走代理。

正确写法是把异步逻辑放到独立的Service里，通过注入调用：

@Service public class QaTaskService { @Async("aiTaskExecutor") public CompletableFuture<String> submitLlmTask(String question) { // 实际的模型调用逻辑 return CompletableFuture.completedFuture(llmClient.call(question)); } }

这里还有个细节：如果方法返回值是void或者普通对象，@Async会把返回值丢弃，想要拿到异步结果并支持扩展，一定要声明为CompletableFuture<T>。

线程池配置是最容易被忽略的。Spring Boot如果不开任何配置，@Async默认用的是SimpleAsyncTaskExecutor，它是不复用线程的，每个任务都新建线程，高并发下能直接创建出几万个线程，堆内存和GC一起崩。实际项目里要显式定义一个线程池Bean：

@Bean("aiTaskExecutor") public ThreadPoolTaskExecutor aiTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(50); executor.setQueueCapacity(1000); executor.setThreadNamePrefix("ai-task-"); executor.setWaitForTasksToCompleteOnShutdown(true); executor.initialize(); return executor; }

参数不是抄来的，要根据下游接口的吞吐和Redis、数据库等资源综合定。核心线程数建议按“单AO接口QPS乘平均耗时”估算，比如目标并发处理200个在途请求，平均耗时5秒，那么至少需要40到50个工作线程才可能满足。

2.3 响应式编程WebFlux：适合SSE流式和大规模IO密集场景

WebFlux是另一条路线。它基于Netty的Reactor，全链路非阻塞，用Mono和Flux表示异步流。AI应用里最常见的SSE（Server-Sent Events）流式输出，用WebFlux做得非常自然——模型生成的token一个接一个推给前端，传统的Servlet模型做流式反而别扭。

@PostMapping(value = "/v1/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux<ServerSentEvent<String>> chat(@RequestBody ChatRequest request) { return llmService.streamResponse(request.getPrompt()) .map(token -> ServerSentEvent.builder(token).build()); }

但WebFlux不是银弹，它要求整个调用链都非阻塞，否则还是会卡线程。如果项目组对响应式编程不熟，WebClient、R2DBC、非阻塞Redis这些中间件都要适配，学习成本和排查成本比异步Servlet高得多。我的看法是：团队里有Reactor经验、或者确实需要长连接流式服务，可以考虑；如果只是为了“异步化”这一个目标，优先选CompletableFuture加虚拟线程或消息队列。

2.4 虚拟线程：JDK 21带来的高并发新解法

虚拟线程是JDK 21正式引入的。它的核心思路是：不再让操作系统线程承载Java线程的所有生命周期，虚拟线程挂起时只占几十KB内存，平台线程空闲后立刻去执行别的虚拟线程。通俗说，以前一个并发请求占一个1MB栈空间的线程，现在只占一点轻量级调度单元，阻塞不再是罪过。

虚拟线程在AI场景确实非常舒坦。以前同步代码要改异步，逻辑被拆得七零八落。虚拟线程允许你继续写同步风格代码，底层却把阻塞挂起释放给平台线程。比如下面的代码，按理说是个同步阻塞调用，但在虚拟线程里，模型等待期间平台线程并没有被占着：

ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor(); public void handleRequest(String question) { virtualExecutor.submit(() -> { String context = retrievalService.search(question); // 阻塞但轻量 String answer = llmService.generate(context); // 阻塞但轻量 notifyClient(answer); }); }

要注意虚拟线程也不是万能的。它不适合CPU密集任务，也不适合带Synchronized块内执行阻塞IO的情况，因为虚拟线程遇到synchronized会锁定载体线程，出现“固定”现象，阻塞照样占住平台线程。此外，虚拟线程不应该池化，它是轻量到“用完即扔”的，搞一个虚拟线程池反而没意义。

2.5 消息队列：异步生产消费模式，扛流量利器

异步化不只是线程模型改造，还包含任务队列。我强烈建议AI应用中，凡是需要长时间处理的任务都考虑消息队列，比如RocketMQ或Kafka。

用消息队列处理AI请求的模式大概是：

用户请求 -> Controller快速响应“已提交” -> 消息发到队列 -> AI消费组拉取消息 -> 调模型 -> 结果写入结果表/推送SSE

这种模式天然削峰填谷。模型服务能力有限，队列不用担心消费者处理不过来，消息都会积压，不会直接丢弃。而且多个消费者实例可以水平扩展，后端加的机器越多，处理能力越强，这个在云原生环境下很友好。

代价是链路变长了，带来额外的消息中间件运维成本和至少几十毫秒的投递延迟。只有业务能接受异步化的，比如工单处理、资料分析、报告生成这类非实时场景，再用队列。如果用户必须立刻拿到完整响应，那还是用同步加异步线程配合限流。

2.6 方案选型速查：一张表把场景对应清楚

选型没有标准答案。我的习惯是：默认用虚拟线程处理阻塞调用，用CompletableFuture编排并行的子步骤，如果场景确实需要流式，再用WebFlux或者Servlet异步加SSE，最后把重任务放到队列去削峰。整套组合下来，既兼顾开发效率，也扛得住实际流量。

3. 高并发设计的四个关键环节

3.1 限流：保护自己，更是保护下游AI服务

高并发系统第一个要求是“活下来”，不是“接得下”。限流就是给系统装刹车。尤其AI场景，下游模型服务往往按Token计费，有QPS配额，你不做限流，模型供应商先把你的建联断掉，整个服务直接雪崩。

限流可以放在网关层，用Sentinel或自研拦截器实现令牌桶算法。令牌桶的好处是允许突发流量，只要桶里有令牌就能放行，而不是死板地平均分配。

@Component public class AiRateLimiter { private final RateLimiter rateLimiter = RateLimiter.create(100); public boolean tryAcquire() { return rateLimiter.tryAcquire(); } }

真正上线前要估算清楚限流阈值。比如目标QPS 100，平均一次模型调用消耗5秒，那么同时挂在系统中的请求就是500个，加上超时和重试的放大，实际要预留至少1.5倍余量。阈值设太低了，业务投诉；太高了，下游被打挂。所以限流要结合压测数据动态调整。

另一个细节是区分“普通用户”和“VIP用户”的配额。用同一个限流器会有饥饿问题，大流量用户把令牌抢光，付费用户进不来。实际做法是分桶限流，每个用户组独立配额，整体并发再设一个总闸。

3.2 熔断与降级：AI服务挂了，你的服务不能跟着挂

熔断和限流是两回事。限流是防止流量过大，熔断是防止下游故障向上游蔓延。我在实际项目里遇到过一个典型：模型供应商服务不稳定，响应偶尔超时到30秒，结果我的服务线程全被这些慢调用占住，后续正常请求也跟着超时。这就是故障放大效应。

用Resilience4j做熔断非常直接。它支持基于失败率、慢调用率来打开熔断器，打开后直接走降级逻辑，不再实际调用下游。

@Bean public CircuitBreaker llmCircuitBreaker() { CircuitBreakerConfig config = CircuitBreakerConfig.custom() .failureRateThreshold(50) .waitDurationInOpenState(Duration.ofSeconds(30)) .permittedNumberOfCallsInHalfOpenState(10) .build(); return CircuitBreaker.of("llm", config); }

熔断后的降级方案要提前想好。能做缓存就返回缓存结果，没有缓存就返回一个通用的“服务暂时繁忙”响应，绝不能让用户请求一路挂着等到超时。这里有个经验之谈：熔断器和线程池隔离要配合使用，每个下游依赖一个独立的线程池，A服务慢了占满A的线程池，最多影响A，不会拖垮主线程池。

3.3 缓存：LLM输出也有大量可复用的部分

很多人觉得AI结果是“随机生成”的，不能缓存。这个想法不全对。实际业务里，FAQ类问题、政策咨询、产品介绍等大量答案是高度相似甚至重复的。我在一个客服项目里做过统计，两周内的高频问题重复率达四成，缓存命中能显著降低模型调用量和用户等待时间。

LLM缓存分两层。一是语义缓存，通过向量相似度判断两个问题是否表达同一个意思，命中后直接返回之前的答案；二是精确匹配缓存，针对完全相同的输入，直接走Redis。

public AnswerResult getCachedAnswer(String question) { String md5Key = DigestUtils.md5DigestAsHex(question.getBytes(StandardCharsets.UTF_8)); String cached = redisTemplate.opsForValue().get("ai:answer:" + md5Key); if (cached != null) { return JsonUtils.parse(cached, AnswerResult.class); } return null; }

但缓存不是全无代价。模型输出可能有随机性，业务方如果要求答案要带时效性（比如价格、政策），必须控制缓存TTL，设置成5分钟或更短。另外，敏感问题不建议缓存，涉及用户个人信息或者合规风险的，一律直连模型，防止缓存穿透或者数据泄露扩散。还有缓存击穿风险——某个高频问题过期瞬间大量回源，解决办法是加互斥锁，只让一个请求去重新生成答案，其余请求等锁后读取新缓存。

3.4 超时与重试：把“坏请求”的影响半径缩到最小

高并发设计里，超时是最容易被低估的参数。很多人不设置HTTP客户端超时，或者设置了一个很大的值，比如60秒，然后抱怨线程不够用。事实上，下游模型服务一旦异常，响应时间会从正常的5秒变成几十秒，你要是不设超时，线程就被烂请求无限期占住。

我见过一份代码里Feign的配置，readTimeout给的是10秒，看起来也不算离谱，但下游偶尔抖动到30秒，Feign不会自动放弃，线程照样干等。正确的做法是先设保守超时——比如模型调用3秒无响应就中断——然后通过重试去弥补偶发失败。超时时间要根据实际响应分布来定，取P99耗时再上浮一小段，不是拍脑袋。

spring: cloud: openfeign: client: config: llm-client: connectTimeout: 2000 readTimeout: 5000

重试也不是越多越好。每次重试都会重新调用下游，如果你的超时是5秒，一次请求最多重试3次，最坏情况一个请求要花15秒等待。高并发下这些重试流量叠加，保护性限流就失效了。重试策略要配合幂等设计，模型调用不幂等，恢复结果会发生重复计费、重复推送。我的经验是：只有明确知道上次失败发生在“网络层”而不是“业务层”时才重试；而且要做退避，第一次等500毫秒，第二次等2秒，减轻下游压力。

4. 实操记录：AI问答平台的高并发后端改造全流程

4.1 业务场景和初始状态

项目背景是给一个企业内部知识库做AI问答平台。员工提问，后端要完成三件事：向量检索相关知识片段、把片段拼接进提示词、调用大模型生成回答。初始实现是全同步的，Controller里直接串行调用三个服务，测试环境一个人用没问题，全网推广当天就崩了。

崩溃现场很典型：Tomcat线程池被打满、模型服务日志全是超时、Redis连接池报错、前端页面大面积“服务暂时不可用”。我从监控里抓到的数据是：Tomcat最大线程200，高峰期在途请求超过230，线程池出现拒绝执行，应用因为OOM差点挂掉。

4.2 改造方案：同步改异步，任务入列，结果回调

我们最后采用的是“同步接受 + 异步处理 + SSE推送”的组合，没有用WebFlux重写整个项目，主要是因为团队对Reactor不熟，迁移成本太高。

具体链路是：

用户POST问题 -> Nginx -> Spring Boot Controller快速返回请求ID -> 问题写入RocketMQ（业务队列） -> AI消费组拉取消息 -> 向量检索并行于大模型生成 -> 结果写入Redis并使用SSE推送给前端。

整套改造花了不到两周，但效果非常明显。在线人数翻三倍，QPS从30支撑到了180，模型调用成功率也从88%提升到了接近99%。

4.3 核心代码片段：从Controller到消费端

Controller只负责接收和快速返回：

@PostMapping("/api/qa") public ResponseEntity<QaSubmitResponse> submit(@RequestBody QaQuestion request) { String requestId = IdGenerator.generate(); qaMessageProducer.send(requestId, request); return ResponseEntity.accepted() .body(new QaSubmitResponse(requestId, "已受理")); }

消息生产者把问题投递到队列：

@Component public class QaMessageProducer { private final RocketMQTemplate rocketMQTemplate; public void send(String requestId, QaQuestion question) { QaMessage message = new QaMessage(requestId, question); rocketMQTemplate.convertAndSend("qa-task-topic", message); } }

消费端拉取消息，执行异步链路：

@Component public class QaTaskConsumer { private final ExampleDataService searchService; private final LlmClient llmClient; private final SsePublisher ssePublisher; public void handle(QaMessage message) { RequestContextHolder.setRequestId(message.getRequestId()); try { // 并行执行检索和生成，缩短总耗时 CompletableFuture<List<String>> searchFuture = CompletableFuture.supplyAsync( () -> searchService.search(message.getQuestion()), searchExecutor); CompletableFuture<String> generateFuture = CompletableFuture.supplyAsync( () -> llmClient.generate(message.getQuestion()), llmExecutor); CompletableFuture<Void> combined = searchFuture .thenAcceptBoth(generateFuture, (contexts, answer) -> { AnswerResult result = new AnswerResult(contexts, answer); ssePublisher.publish(message.getRequestId(), result); }); combined.orTimeout(30, TimeUnit.SECONDS) .exceptionally(ex -> { log.error("qa task failed", ex); ssePublisher.publishError(message.getRequestId(), "生成失败，请稍后重试"); return null; }); } finally { RequestContextHolder.clear(); } } }

注意我在这里用了orTimeout，这个API从Java 9开始提供，它能在指定时间后主动完成Future并抛异常，是防止任务无限卡死的利器。异步链路上不设超时，等同于同步代码里不设readTimeout，早晚出事。

4.4 线程池与队列参数怎么定

这是改造里最需要抠细节的部分。我们最后给检索和大模型分别建了线程池。

检索线程池：

• 核心线程15，最大线程30，队列容量200

因为向量检索平均耗时80ms，P99大概200ms，扛住每秒钟几百次检索绰绰有余。

模型线程池：

• 核心线程20，最大线程40，队列容量500

大模型平均耗时3秒，P99可能要8秒，所以线程数反而要多。估算逻辑是：目标在线处理500个模型任务，平均耗时3秒，至少需要500除以1约等于500个线程才能保证每秒处理166个任务？这个算错了，我想想。

准确计算：如果每秒进入500个模型调用，每个耗时3秒，那么系统同时处于在途状态的调用数量大约是500乘以3等于1500个。线程数不可能配1500，那意味着要控制下游接受能力，同时要依赖限流把入口的QPS降下来。实际业务里我们不追求单个节点吃下所有流量，而是QPS控制到模型能力之内，然后横向扩容节点数。

线程池的队列容量也要算。QueueCapacity填500，意味着线程池最多接受40加500共540个任务，多出来的直接拒绝。拒绝策略我选了CallerRunsPolicy，让提交任务的线程自己执行一个兜底逻辑，把任务再由生产者发回队列，防止数据丢失。

4.5 限流、熔断和数据一致性收尾

在网关层加了Sentinel限流，按用户ID分桶，普通用户每人每秒钟最多3个问题，VIP用户可以放宽到10个。全局最大QPS设置为150，超过后返回友好的“系统繁忙，请稍候再试”。下游模型服务专门加了一个Resilience4j熔断，60秒内失败率达到50%，熔断30秒。

数据一致性这块常被忽略。异步结果写回Redis时，如果用户已经关闭浏览器，SSE连接断了，消息不能默默丢弃，我选择把结果持久化到一张结果表，前端下次刷新时可以按requestId拉取历史答案。这样即使某个环节失败，用户再点一次也能拿到结果。

压测结果比较理想。使用JMeter模拟200并发用户，每用户连续提问5轮，平均响应时间（SSE首包时间）从原来的4000ms降到了700ms，请求成功率从88%提升到99.6%。整体QPS在4台4核8G的节点上跑到180，再往上提升就得加节点了。

5. 实战常见的6个问题和排查技巧

5.1 线程池满了，报“Task rejected”该查什么

这个报错是最常见的。先看线程池核心线程、最大线程、队列容量三个参数匹配不匹配。很多人只调大了最大线程数，但队列容量设了100000，等于永远触达不到最大线程，任务全积压队列里，表面线程没满，实际响应延迟已经拉垮了。

Tomcat的线程池队列默认是无限长，这是个大坑。Tomcat最大线程200，如果队列无限积压，请求不会返回报错，但每个请求的等待时间不断增长，用户点击后要卡几十秒才响应。这种情况体现出来的问题不是“线程满了”，而是“线程没有被释放”。排查时要同时看活跃线程数、队列积压数和下游响应时间，三个指标放一起才能定位是哪里慢。

5.2 CompletableFuture用了默认线程池，全链路拖垮

项目初期有人直接用CompletableFuture.runAsync()，没传线程池，结果用的是ForkJoinPool.commonPool。这个池的线程数等于CPU核数减1，AI场景下几十个并发任务就排队等线程，性能还不如同步执行。

排查方法很简单：看线程Dump里有没有大量名为ForkJoinPool.commonPool-worker的线程，如果有，代码里八成漏传了线程池。修复方式是把所有supplyAsync调用改成显式传线程池，并且和业务线程池隔离，各管各的。

5.3 主线程关闭时异步任务被“吞掉”

Spring Boot应用如果直接重启，正在执行的异步任务会被中断，消息队列里还没消费的任务也原地消失。我第一次上线就踩了这个坑：发布新版本前没有任何告警，结果发布完用户发现少了一批回答。

解决办法有两条路。线程池要配置优雅停机，setWaitForTasksToCompleteOnShutdown(true)让任务在关闭前尽量完成，再给一个setAwaitTerminationSeconds(30)上限，避免停机卡死；消息队列消费组要开启断点续传，消费成功后确认位移，重启后从记录点继续消费。

5.4 traceId在异步链路里丢失，排查问题无从下手

同步代码里用日志追踪非常方便，一个请求的日志是一串日志串起来的。但异步化之后，线程发生了变化，ThreadLocal里的traceId不会自动传递，日志就成了碎片，定位一个用户的问题要翻半天。

我的做法是使用TransmittableThreadLocal，或者用Spring Cloud Sleuth这类链路追踪组件，从HTTP入口生成traceId，往消息体里塞一份，消费端再重新绑定上下文。这样从用户提问到模型生成、SSE推送的全过程，日志都能串起来。别小看这一步，等线上出了问题你会感谢当初这十分钟的配置。

5.5 慢消费导致消息积压，下游被压垮

消息队列解耦了流量，但也带来了新的风险：如果消费速度跟不上生产速度，消息积压会越来越大，积压到一定量，下游模型服务可能会被“追债式”流量引爆。这时候光加消费实例不够，要先看看消费端每个任务的平均耗时是不是变长了，可能是模型接口变慢，也可能是下游依赖的服务有抖动。

稳妥做法是对消费组单独做限流，配置每秒钟最大消费数量，让消费速度平缓，不要猛灌。监控上要注意积压条数这个指标，超过阈值就告警。积压不是问题，无感知的积压才是大问题。

5.6 数据竞态：异步结果写到同一个Key，互相覆盖

多个子任务并行后，如果它们同时更新同一个缓存Key或同一个数据库行，就有竞态问题。模型场景里不太明显，但RAG检索片段拼接时，如果多线程往同一个StringBuilder里追加内容，顺序就会错乱。

处理策略是：每个异步子任务只负责处理自己的中间结果，最终拼接放到下一个阶段单独做；如果要并发更新同一个记录，用数据库乐观锁版本号控制；Redis里则用setIfAbsent加锁键来串行化写操作。

写到最后，我想分享点体会。刚开始接触AI应用的时候，我总想着把异步化和高并发做成一套很“炫”的架构，WebFlux、响应式、消息队列全上。后来被线上问题教育了几次才明白，架构是为稳定性服务的，不是为简历服务的。很多项目根本不需要全链路响应式，一套“虚拟线程加CompletableFuture加限流熔断缓存”的组合，已经能覆盖绝大多数场景。

如果在座各位刚接手类似项目，我建议不要一上来就改架构。先打好监控，把线程数、队列积压、下游耗时、限流阈值这些数据捞出来，看清楚瓶颈在哪里，再决定改哪里。每一步改动都做压测对比，用数据说话。这是个慢功夫，但也是最靠谱的路。