尧图网站建设 
如果你的Java服务要对接大模型接口用户每次发来一句话后端就要在那个线程上傻等三到十秒等并发一上来两百个Tomcat线程瞬间被占满新的请求全在队列里排队这种日子我确实经历过。很多人以为高并发优化就是调大线程池、加缓存但AI应用跟普通接口最大的区别是你的系统不是被CPU打垮的是被等待打垮的。一次生成式AI请求既要等远端接口吐token中间可能还要并行做向量检索、多路召回、工具调用链路里全是长耗时IO。这种负载模型下Java传统的同步阻塞编程方式必须转成异步化配合高并发设计。这篇就把我在实践中整理的异步化工具选型、高并发防护手段、以及一个可复现的AI Agent骨架全部分享出来。先说结论AI应用的异步化核心不是加快单个请求而是把占着茅坑不拉屎的线程释放出来。传统Web接口的RT响应时间是几十毫秒AI接口的RT往往是几秒到几十秒差了整整两个数量级。同样的200个线程传统接口能扛几千QPSAI接口可能扛50 QPS都费劲。所以这篇文章适合三类人正在做AI应用后端开发的工程师、准备面试Java高并发岗位的开发者、以及想从传统Web转AI方向但心里没底的同学。1. 为什么AI应用必须打破一个请求一个线程的老规矩1.1 AI请求不是短事务而是长尾阻塞传统Java Web应用的标准模型是一个请求占用一个线程处理完就释放。Servlet容器Tomcat、Jetty内部维护了一个线程池默认情况下200个线程左右。传统业务里一次请求经历数据库查询、Redis读取、远程RPC调用总耗时通常控制在50毫秒以内线程被占用的时间非常短200个线程可以循环复用支撑几千甚至上万QPS没问题。但AI应用完全不同。当你调用一个云端大模型接口时用户发出对话请求后模型需要先解析Prompt再逐token生成回复文案一个普通对话请求的响应时间通常在3到10秒复杂推理场景能到30秒以上流式输出场景下连接要维持几秒到几十秒期间线程持续被占用如果你做了RAG检索增强生成还得先去向量数据库召回文档、再拼Prompt这意味着什么假设你的AI接口平均RT是5秒Tomcat默认200个线程的极限吞吐就是200÷540 QPS。一旦请求量稍微上来线程池直接耗尽后面的请求全部在队列里等待用户体感就是转圈圈转半天。更可怕的是长尾效应——AI模型的响应时间方差极大P99可能到了15秒这些慢请求会长期霸占线程把整个服务的吞吐拖垮。我自己踩过的坑早期做一个问答系统后端直接同步调外部大模型接口压测的时候40并发就把服务打崩了。jstack一看两百多个Tomcat线程全部BLOCKED在HTTP连接上一个都不剩。那时候才意识到同步阻塞模型在AI场景下不是性能问题而是架构问题。1.2 异步化的本质把等待变成可让出的资源异步化的本质其实很朴素线程不应该傻等IO完成。传统的同步模型里线程发起一个HTTP调用后自己进入阻塞状态直到响应返回才继续执行。这段时间线程什么事都没干纯粹是占着内存等网络包。异步化的思路就是发起调用后线程立即被释放回池子里干别的活等数据到达了再通过回调或者挂起恢复来处理。在AI场景这个特性尤其关键。因为AI接口调用中有大量的网络等待、模型排队、token生成耗时实际CPU计算量反而很低。举个例子一次单轮对话请求后端调LLM花了8秒其中纯CPU计算可能只有几百毫秒剩下7秒多是等网络、等对方的GPU排队。如果你用同步模型8秒内这个线程全程被占用。如果用异步化或者虚拟线程模型这8秒内线程资源可以被几十个其他任务复用系统的整体吞吐能力自然就上去了。但这里要澄清一个误区异步化并不会降低单个请求的RT它优化的是单位时间内能处理的请求数量也就是吞吐。用户该等3秒还是等3秒但同样硬件条件下系统能同时服务更多人。理解了这一层再去看各种异步化工具思路就清晰多了。2. Java异步化三件套怎么配合线程池、CompletableFuture、虚拟线程2.1 线程池把等待时间变成资源预算Java里最经典的异步手段就是线程池。AI应用里线程池大小不能像传统Web那样拍脑袋定它有明确的公式线程数 目标QPS × 单请求RT秒假设你的业务目标是支撑50 QPS接口平均RT为5秒那么需要的同时在途线程数就是50 × 5 250。这250个线程不应该放在Tomcat容器线程里而应该放在独立的业务IO线程池中。为什么因为容器线程还要负责接收HTTP连接、做参数校验、返回响应等基础工作如果被AI长任务占满连健康检查接口都响应不了。实际配置时我通常这样定义一个专用线程池ThreadPoolExecutor aiExecutor new ThreadPoolExecutor( 128, // 核心线程数按目标QPS × RT估算 256, // 最大线程数留出峰值余量 60L, TimeUnit.SECONDS, new ArrayBlockingQueue(500), // 有界队列防止内存被耗尽 new ThreadFactoryBuilder().setNameFormat(ai-invoke-%d).build(), new ThreadPoolExecutor.AbortPolicy() // 队列满了直接拒绝快速失败 );两个关键点队列必须用有界队列否则请求无限堆积会吃掉所有堆内存拒绝策略用AbortPolicy让过载请求快速失败返回错误码而不是在队列里熬到超时。线程名一定要用有意义的名称排查线上问题时jstack一眼就能看出哪些线程在忙什么。线程数的确定不是一劳永逸的要根据压测结果持续调整。我见过有人把线程池调到4096结果下游AI接口每秒最多处理20个请求大量线程全部卡在等待响应的路上内存占用飙升毫无意义。线程池的大小最终由下游瓶颈决定不是越大越好。2.2 CompletableFuture编排并发分支别把回调写成回调地狱线程池解决了谁来干活的问题但AI应用里还有活怎么编排的问题。一次用户提问后端可能要同时做几件事判断缓存是否有命中从向量数据库检索相关文档从搜索引擎或ES做关键词召回组装Prompt后调用LLM这些分支有的可以并行有的要合并结果后再继续。用传统Future加回调写出来的代码很快会陷入回调地狱。CompletableFuture是Java 8引入的异步编排利器它把异步任务抽象成可以组合的流水线用起来比手写线程同步舒服得多。CompletableFutureListDocument vectorFuture CompletableFuture .supplyAsync(() - vectorStore.search(question, 5), retrievalExecutor); CompletableFutureListDocument keywordFuture CompletableFuture .supplyAsync(() - searchEngine.search(question, 5), retrievalExecutor); CompletableFutureListDocument docsFuture vectorFuture .thenCombine(keywordFuture, this::mergeAndDeduplicate) .orTimeout(3, TimeUnit.SECONDS) // 检索最多等3秒 .exceptionally(ex - List.of()); // 检索失败降级为空列表这一步把两路检索并行跑最慢的一路决定了整体等待时间而不是串行累加。thenCombine把两个结果合并成一路orTimeout给整个检索过程设了上限exceptionally保证即使检索全挂了也不会让整个请求抛异常而是按空文档继续走。这个类有三个坑必须记牢。第一个坑默认使用ForkJoinPool.commonPool。如果你写的supplyAsync没有传第二个参数任务会提交到JVM全局共享的ForkJoinPool它的线程数默认是CPU核心数-1。AI场景下大量任务都在阻塞等待网络IO这点线程根本不够用一旦占满所有异步任务全部排队。任何耗时调用都必须显式传入自定义线程池回调里的子任务也一样。第二个坑异常容易丢失。CompletableFuture链路上任何一个环节抛异常如果后续没有exceptionally或handle兜底异常会被吞掉表现为请求莫名其妙超时日志里啥也没有。所有异步链路末尾必须挂兜底异常处理。第三个坑回调线程不可控。thenApply默认在触发它的那个线程执行thenApplyAsync才会重新提交到线程池。如果你依赖某个回调一定在哪个线程跑来关联上下文比如传递TraceId记得用thenApplyAsync指定线程池。2.3 虚拟线程JDK 21带来的伪同步写法如果说CompletableFuture解决的是编排问题那虚拟线程解决的是写代码的体感问题。JDK 21正式发布了虚拟线程它为大量阻塞任务而生。虚拟线程由JVM调度占用内存极小默认栈只有十几KB而平台线程默认1MB左右创建成本几乎可以忽略不计。当一个虚拟线程执行阻塞IO时它会自动释放底下承载它的平台线程载体线程让载体线程去跑别的虚报线程。这意味着你可以用同步阻塞的代码风格享受到异步化吞吐的好处。不用CompletableFuture不用回调直接写成// 定义一个虚拟线程池随用随建用完就扔 ExecutorService aiExecutor Executors.newVirtualThreadPerTaskExecutor(); // 业务代码里直接用同步风格 public String chat(String question) throws Exception { FutureString llmFuture aiExecutor.submit(() - llmClient.call(question)); FutureListDocument docFuture aiExecutor.submit(() - vectorStore.search(question)); String answer llmFuture.get(10, TimeUnit.SECONDS); ListDocument docs docFuture.get(3, TimeUnit.SECONDS); return buildResponse(answer, docs); }aiExecutor.submit配合Future.get代码看着像同步但每个阻塞调用都不占平台线程几十万个虚拟线程并发也是常态。这就是JDK 21带来的最大改变你不需要刻意学习响应式编程也能写出高吞吐的IO密集应用。但虚拟线程有三个禁区要避开。第一CPU密集型任务不要用虚拟线程它不会让CPU变快反而增加调度开销。第二别在synchronized块里做阻塞IOsynchronized是平台线程级别的锁虚拟线程一旦在锁内阻塞会钉住载体线程。第三避免使用JNI方法同样会钉住载体线程。AI应用里大量的是HTTP调用、向量检索这类IO操作放心用虚拟线程没问题。实际生产里我不会只用一种方案。通常的做法是用CompletableFuture做多路分支的编排用虚拟线程池做实际阻塞IO的落地用平台线程池或Tomcat容器线程接收请求和快速响应。各取所长。3. 高并发AI应用的拦截与保护层设计异步化把系统吞吐拉开之后新的问题来了你这边吞吐上去了下游AI接口能接住吗即使能接住你的内存和上游配额扛得住吗高并发不只是把请求发出去更重要的是发不出去的时候怎么办。这一层我把它叫保护层是整个高并发设计的重中之重。3.1 限流信号量和令牌桶到底该用哪个AI应用的限流有两条路控制同时在途的并发数和控制单位时间的请求速率。并发数控制最直接的工具是Semaphore。你的上游AI账号可能只允许5个并发请求那就在服务里声明一个Semaphore(5)每次要调AI接口前先tryAcquire()拿到了就调拿不到就排队或返回降级提示。这比请求进来后再被上游限流明智得多因为上游的一次429响应也是要花时间的。private final Semaphore llmConcurrencyLimiter new Semaphore(5); public String callLlm(String prompt) { if (!llmConcurrencyLimiter.tryAcquire(2, TimeUnit.SECONDS)) { throw new TooManyRequestsException(系统繁忙请稍后再试); } try { return llmClient.call(prompt); } finally { llmConcurrencyLimiter.release(); } }速率控制则用令牌桶思路。比如限制每秒最多进来20个请求用Guava RateLimiter或者自研一个简单的令牌桶都行。AI场景比较特殊我建议两种都做入口对全局限速防止流量洪峰打垮服务调用上游前对单账号并发限流防止被上游封掉账号。令牌桶和信号量的选型标准我总结成三句话如果你关心同一时刻最多多少人调用用信号量如果你关心每秒最多处理多少请求用令牌桶如果两者都关心先用令牌桶挡流量再用信号量保上游。3.2 超时、熔断与重试防连环雪崩的三道闸门高并发系统最怕的是一个慢请求拖垮全链路。AI接口一旦上游波动响应时间从2秒飙到60秒你这边如果无限期等待线程池很快就被这些慢请求占满随后所有新请求全部遭殃。所以三道闸门必须全部做。第一道闸门是超时。每一条AI调用都要设置连接超时、读取超时整个链路还要设置总预算。我用HTTP客户端举例HttpClient client HttpClient.newBuilder() .connectTimeout(Duration.ofMillis(800)) // 连不上赶紧放弃 .executor(aiExecutor) // 使用虚拟线程池 .build(); HttpRequest request HttpRequest.newBuilder(uri) .timeout(Duration.ofSeconds(15)) // 整个请求最多15秒 .POST(...) .build();注意timeout这个参数是整个请求从开始到读完响应的总时长但流式输出场景里只要持续拿到数据就不应该中断这需要按数据块来中断判断不能拿总超时一刀切。这里我用HttpClient的BodyHandlers.ofLines()配合Flow API逐行处理每行之间的间隔超过阈值就视为断流。第二道闸门是熔断。熔断的意思不是超时了重试一次而是连续失败多了直接进入拒绝状态不再发请求给上游。像Resilience4j这类库可以直接集成但AI场景我也见过手写熔断器的核心参数就是三个窗口大小统计最近N次请求、失败率阈值比如超过50%就熔断、恢复时间熔断后过多久放一个试探请求。AI模型上游不稳定是常态熔断能保护你的线程池和余额。// 一个极简熔断器统计最近20次调用 class LlmCircuitBreaker { private final int windowSize 20; private final double failureThreshold 0.5; private final AtomicLong windowStart new AtomicLong(); private final AtomicInteger requestCount new AtomicInteger(); private final AtomicInteger failureCount new AtomicInteger(); private volatile boolean open false; public boolean canPass() { if (open) { // 熔断状态下每5秒放一个试探请求 if (System.currentTimeMillis() - windowStart.get() 5000) { return true; } return false; } return true; } public void recordResult(boolean success) { ... } }第三道闸门是重试。这个在AI场景要额外的谨慎。大模型接口不是幂等的你重试一次它重新生成一次账单一分不少地扣。重试策略必须满足三个条件请求确实是网络中断或超时不是业务失败你有请求ID可以传给上游做去重重试间隔是指数退避1秒、2秒、4秒、最大10秒并且总的额外等待时间不超过你的链路预算。我的实践原则是宁可返回暂时不可用也不要为了成功率高而乱重试。AI服务的高价值请求可以用一次重试兜底普通对话请求重试一次就够了。3.3 背压与内存保护别让队列吃掉你的堆内存异步化系统有一个隐蔽的杀手——任务堆积。当请求进来的速度大于处理速度时所有未处理的任务都会堆积在队列里。如果队列是无界的最终结果就是把堆内存填满触发Full GC然后整个应用卡死。有界队列是第一步前面线程池示例里我用ArrayBlockingQueue(500)就是干这个的。第二步是入口信号量在请求进入业务层之前就做一次并发拦截防止瞬间洪峰把所有队列都塞满private final Semaphore requestLimiter new Semaphore(100); public void handleRequest(HttpServletRequest request, HttpServletResponse response) { if (!requestLimiter.tryAcquire()) { // 直接返回繁忙不进入后续处理 response.setStatus(503); return; } try { processAsync(request, response); } finally { requestLimiter.release(); } }第三步是考虑AI应用特有的内存压力大模型的输入输出token是有长度的。你在入口不做长度限制一个用户提交了5万字的问题你把全文塞进Prompt然后转发给模型这个请求占的内存、网络带宽、上游计费都是普通请求的几十倍。高并发场景必须对Prompt长度做限制超长的要么截断要么拒绝。另外响应结果的缓存比传统接口更值得做。同一个热门问题完全可以让后续用户直接命中缓存不再重复调用大模型。我见过一个客服问答系统加了Embedding相似度缓存之后上游调用量直接降了60%。缓存的是向量检索结果LLM回复用相似度阈值判断是否命中。4. 实操一个可复现的AI流式问答Agent骨架理论讲再多不如直接看一个能跑起来的骨架。下面这个示例我尽量保持代码简洁但保留了完整的高并发设计要点异步编排、虚拟线程、信号量限流、SSE流式输出、断连取消。4.1 整体架构与并发模型设计先描述一下目标场景Web页面提供一个对话框用户输入问题后后端需要做两路检索向量库ES再把检索结果拼进Prompt最后以流式方式把大模型的回复逐步推送到浏览器。要求是首字到达尽量快用户不用等所有token生成完才看到内容高并发下系统不死过载时快速失败客户端中途关闭页面后端要立刻停止调用大模型省钱并发模型设计如下Tomcat容器线程只负责接收HTTP请求、创建SSE连接、把任务丢给业务线程池后立即返回检索和LLM调用都在虚拟线程池里执行用CompletableFuture做两路检索的合并编排用有界队列信号量保护系统不被打垮。这里用SseEmitter它是Spring MVC原生支持的服务端推送对象用法简单兼容性好。配合虚拟线程池代码风格几乎是同步的写起来非常轻松。4.2 Controller层快速返回不做重活RestController RequestMapping(/api) public class ChatController { private final ChatService chatService; GetMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter chat(RequestParam(question) String question) { SseEmitter emitter new SseEmitter(60_000L); // 单次流式最长60秒 chatService.chatAsync(question, emitter); return emitter; } }注意Controller方法本身不做任何耗时操作它创建了SseEmitter然后立即调用chatAsync把工作异步化。这里的chatAsync内部会立刻启动新的虚拟线程去干活方法本身立刻返回Tomcat线程马上空出来接下一个请求。为什么要返回SseEmitter而不是直接等待结果呢因为大模型回答是流式的你可能在4秒后就要推送第一个token但完整的回答需要15秒。SseEmitter让你可以持续地向客户端推送数据而不是等全部完成再一次性返回。4.3 Service层异步编排主流程Service public class ChatService { private final VectorStore vectorStore; private final SearchEngine searchEngine; private final LlmClient llmClient; // 虚拟线程池专门跑阻塞IO任务 private final ExecutorService ioExecutor Executors.newVirtualThreadPerTaskExecutor(); // 检索结果合并线程池也可以复用上面的 private final ExecutorService mergeExecutor Executors.newVirtualThreadPerTaskExecutor(); // 限制同时调用大模型的并发数账号配额是5并发 private final Semaphore llmConcurrencyLimiter new Semaphore(5); // 入口限流防止服务过载 private final Semaphore entranceLimiter new Semaphore(100); /** * 异步处理聊天请求结果通过 SseEmitter 持续推送 */ public void chatAsync(String question, SseEmitter emitter) { if (!entranceLimiter.tryAcquire()) { emitter.completeWithError(new TooManyRequestsException()); return; } // 在虚拟线程池里跑整个流程Controller 立即返回 ioExecutor.submit(() - processChat(question, emitter)); entranceLimiter.release(); } private void processChat(String question, SseEmitter emitter) { try { // 1. 两路检索并行发出去 CompletableFutureListString vectorFuture CompletableFuture.supplyAsync( () - vectorStore.search(question, 5), ioExecutor); CompletableFutureListString keywordFuture CompletableFuture.supplyAsync( () - searchEngine.search(question, 5), ioExecutor); // 2. 合并检索结果最多等3秒失败降级为空列表 CompletableFutureString promptFuture vectorFuture .thenCombine(keywordFuture, (docs, keywords) - buildPrompt(question, concat(docs, keywords))) .orTimeout(3, TimeUnit.SECONDS) .exceptionally(ex - { log.warn(retrieve failed, fallback to direct chat: {}, ex.toString()); return buildPrompt(question, ); }); // 3. 拿到Prompt后流式调大模型 promptFuture.thenAcceptAsync(prompt - { streamLlmAndSend(emitter, prompt); }, ioExecutor); } catch (Exception e) { log.error(chat failed, e); emitter.completeWithError(e); } } }这个流程里有几个值得琢磨的设计点。第一thenAcceptAsync执行的时候已经拿到了完整Prompt它内部会再次在虚拟线程池里发起大模型流式调用。第二为了避免用户在检索阶段干等理想情况下应该在大模型能接受的条件下尽早把首字推出来实际项目里甚至可以把检索结果也做成流式推给前端展示。第三entranceLimiter用来挡住过载流量llmConcurrencyLimiter号在streamLlmAndSend内部再获取两者负责不同角色的限流。这段代码没有用Async注解因为Async底层是代理线程池配置起来绕来绕去不如直接显式声明ExecutorService一目了然。排查问题时能直接看到线程名知道任务跑在哪个池子里。4.4 流式输出与断连取消最容易被忽略的环节大模型流式调用的典型实现是用HTTP客户端读流拿到每个增量token就通过SseEmitter发出去。伪代码如下private void streamLlmAndSend(SseEmitter emitter, String prompt) { if (!llmConcurrencyLimiter.tryAcquire()) { emitter.send(SseEmitter.event().data(当前请求过多请稍后再试)); emitter.complete(); return; } try { HttpClient client HttpClient.newBuilder() .connectTimeout(Duration.ofMillis(800)) .executor(ioExecutor) // 重要用我们的虚拟线程池 .build(); HttpRequest request HttpRequest.newBuilder() .uri(URI.create(https://api.example.com/v1/chat)) .timeout(Duration.ofSeconds(30)) .header(Authorization, Bearer xxx) .header(Accept, text/event-stream) .POST(BodyPublishers.ofString({\prompt\:\ escapeJson(prompt) \,\stream\:true})) .build(); // 注册SseEmitter的断开回调一旦客户端断开就取消请求 AtomicBoolean cancelled new AtomicBoolean(false); emitter.onCompletion(() - cancelled.set(true)); emitter.onTimeout(() - cancelled.set(true)); client.sendAsync(request, BodyHandlers.ofLines()) .thenAcceptAsync(response - { response.body().forEach(line - { if (!cancelled.get()) { try { String token parseSseToken(line); if (token ! null) { emitter.send(SseEmitter.event().data(token)); } } catch (Exception e) { // 客户端断开会抛 IOException这里必须终止 cancelled.set(true); return; } } if (cancelled.get()) { // 让上游断开流 throw new CompletionException(new IOException(client cancelled)); } }); }, ioExecutor) .exceptionally(ex - { log.warn(stream interrupted or failed: {}, ex.getMessage()); return null; }) .thenRun(() - { llmConcurrencyLimiter.release(); emitter.complete(); }); } catch (Exception e) { llmConcurrencyLimiter.release(); emitter.completeWithError(e); } }这段代码有几个细节很关键。onCompletion和onTimeout回调里不能直接阻塞式地关闭HTTP连接所以我用AtomicBoolean做标志在下一次处理token时检查。emitter.send如果抛异常说明客户端已经断开这时候要把整个流处理链路终止掉否则大模型还在继续生成token每一秒都在烧钱。注意客户端断开后必须调用远端接口的cancel或关闭响应流。有些HTTP客户端库不会自动中断你要在onCompletion回调里显式地response.cancel()。省下来的每一个token都是真金白银。4.5 参数计算实例50 QPS的配置推导光给代码不给参数不算真正可复现。下面我给一个具体场景的完整配置推导过程。假设业务目标峰值50 QPSAI接口平均RT 4秒两路检索平均RT 1秒上游AI账号限制5并发。第一项虚拟线程池大小。50 QPS × 4秒 200个同时在途的AI请求。虚拟线程本身很轻量但也不能完全无脑开我设置newVirtualThreadPerTaskExecutorJVM按需创建只要底层载体线程够用即可。载体线程默认数量与CPU核数相关实际IO等待时大部分会被释放所以提供200个并发虚拟线程没有压力。但注意HTTP连接池、下游连接数都是物理上限虚拟线程再多也不会超过这些瓶颈。第二项检索线程池。50 QPS × 1秒 50个并发检索任务。复用ioExecutor即可注意虚拟线程池创建任务无上限这里真正限制并发的是下游检索服务的连接数所以检索客户端的连接池要能扛住50并发。第三项信号量参数。llmConcurrencyLimiter new Semaphore(5)这是硬限制来自上游账号配额。这里会出现一个现象虚拟线程池允许200个任务但信号量只放5个进AI调用其余195个虚拟线程会阻塞在tryAcquire上。这是完全正常的虚拟线程挂在信号量上内存占比小系统不会崩只是请求排队。第四项入口信号量。目标是快速失败我设置100。超过100个同时在途请求直接返回503用户看到的提示是系统繁忙而不是无限等待。第五项队列与拒绝策略。如果不用信号量而是用线程池队列我建议队列长度设为QPS × 容忍的排队秒数。假设容忍排队2秒队列长度就是50 × 2 100。超过之后AbortPolicy触发快速失败。5. 常见问题与排查技巧实录5.1 线程池被打满从哪里下手查典型症状请求大量超时服务整体吞吐掉到接近零。第一步先jstack看线程状态分布重点看两种大量线程WAITING (parking)说明线程池任务队列在排队或者CompletableFuture在等待结果大量线程RUNNABLE且堆栈停在SocketInputStream.read说明线程都卡在等待网络IO用socketTimeout和连接池上限去排查看线程名很关键这就是前面强调必须自定义线程名的原因。如果线程名都是ai-invoke-%d那就去查这个池子的任务是谁提交的、下游为什么慢。5.2 CompletableFuture默认线程池饥饿有段时间我做RAG检索supplyAsync没传线程池压测30并发时莫名大量超时日志里却没有任何异常。后来用jstack发现commonPool线程全部WAITING在检索调用上但CPU核心数只有8commonPool最多7个线程全被阻塞IO占住后续任务全部排队。这个坑非常隐蔽因为代码看起来完全正常只是超时。检查方法很简单在所有supplyAsync、thenApplyAsync、runAsync调用上加断点看是否有调用没有传Executor参数。或者全局搜一下Async方法逐个确认。这个错误在我团队里至少被踩过三次。5.3 虚拟线程下连接池反而成了新瓶颈虚拟线程可以创建成千上万个但你的数据库连接池、Redis连接池、HTTP连接池还是物理限制。最常见的错误是虚拟线程开了5000个全部去抢数据库连接的信号量结果连接池只有50个4950个虚拟线程全堵在等连接上数据库负载没涨应用倒是先卡死了。解决办法是连接池大小的确定要放在虚拟线程数之前。先算下游并发上限再决定你的虚拟线程规模。比如数据库最大连接数是50那你全局的虚拟线程并发度就别超过几百否则徒增无意义的线程切换和排队。5.4 客户端断开后上游还在持续计费流式场景下用户不会每次都很温柔地等token全部返回。关网页、切应用、刷新页面都意味着SSE连接断开。如果后端没有在onCompletion里取消上游请求模型服务会一直生成到自然结束。我见过真实的账单惨案一个几十人的内测系统一个下午因为断连未取消花了平时半个月的模型调用费用。建议在代码里显式把断连纳入主流程处理而不是当作异常边缘情况。SseEmitter的onCompletion回调里不但要置取消标志最好还能拿到底层HTTP连接的引用直接关闭。5.5 涉及面试官常问的几个点怎么答这个话题经常出现在Java面试题讨论里我踩过的坑基本就是面试官想挖的点。核心问题是AI应用的高并发设计和传统Web高并发有什么异同好的回答思路是先点明负载模型差异长耗时、外部不可控、流式再说线程模型差异同步阻塞不可行然后落到工具选型CompletableFuture编排虚拟线程落地最后提到保护层限流、熔断、超时、背压。面试官会追问CompletableFuture默认线程池有什么坑虚拟线程和平台线程的区别AI接口该不该重试都是一线问题把我上面写的经验答出来基本就够了。6. 最后再分享一点实操体会文章写到这里我把自己在Java AI应用异步化与高并发设计上的沉淀做了个完整梳理但每个系统都有自己特殊的地方。落到自己的项目时请第一个去看你的下游瓶颈在哪里是模型接口并发限制、是向量库性能、还是网络带宽。先确定瓶颈再去设计线程池、信号量和队列顺序反了容易白忙活。我的体会是Java 21之后做AI应用后端技术上的标准答案越来越清晰了虚拟线程负责解决IO密集的吞吐问题CompletableFuture负责解决多路调用的编排问题限流熔断超时负责解决外部依赖不稳的问题。三者各管一段不需要引入复杂的流式编程框架也不需要靠调整堆大小硬扛。最后一个小建议做这种系统的压测时不要只看总QPS。把P50、P95、P99分开看重点观察线程池活跃度、信号量排队数、下游接口拒绝次数这三个指标。只有它们同时处在一个合理范围你的异步化架构才算真正立住了。如果发现P99涨得飞快而P50还好多半是排队问题——这时候不是异步化错了是入口限流或队列容量还需要再调整。 Java AI应用高并发异步架构设计：从线程池到虚拟线程
如果你的Java服务要对接大模型接口用户每次发来一句话后端就要在那个线程上傻等三到十秒等并发一上来两百个Tomcat线程瞬间被占满新的请求全在队列里排队这种日子我确实经历过。很多人以为高并发优化就是调大线程池、加缓存但AI应用跟普通接口最大的区别是你的系统不是被CPU打垮的是被等待打垮的。一次生成式AI请求既要等远端接口吐token中间可能还要并行做向量检索、多路召回、工具调用链路里全是长耗时IO。这种负载模型下Java传统的同步阻塞编程方式必须转成异步化配合高并发设计。这篇就把我在实践中整理的异步化工具选型、高并发防护手段、以及一个可复现的AI Agent骨架全部分享出来。先说结论AI应用的异步化核心不是加快单个请求而是把占着茅坑不拉屎的线程释放出来。传统Web接口的RT响应时间是几十毫秒AI接口的RT往往是几秒到几十秒差了整整两个数量级。同样的200个线程传统接口能扛几千QPSAI接口可能扛50 QPS都费劲。所以这篇文章适合三类人正在做AI应用后端开发的工程师、准备面试Java高并发岗位的开发者、以及想从传统Web转AI方向但心里没底的同学。1. 为什么AI应用必须打破一个请求一个线程的老规矩1.1 AI请求不是短事务而是长尾阻塞传统Java Web应用的标准模型是一个请求占用一个线程处理完就释放。Servlet容器Tomcat、Jetty内部维护了一个线程池默认情况下200个线程左右。传统业务里一次请求经历数据库查询、Redis读取、远程RPC调用总耗时通常控制在50毫秒以内线程被占用的时间非常短200个线程可以循环复用支撑几千甚至上万QPS没问题。但AI应用完全不同。当你调用一个云端大模型接口时用户发出对话请求后模型需要先解析Prompt再逐token生成回复文案一个普通对话请求的响应时间通常在3到10秒复杂推理场景能到30秒以上流式输出场景下连接要维持几秒到几十秒期间线程持续被占用如果你做了RAG检索增强生成还得先去向量数据库召回文档、再拼Prompt这意味着什么假设你的AI接口平均RT是5秒Tomcat默认200个线程的极限吞吐就是200÷540 QPS。一旦请求量稍微上来线程池直接耗尽后面的请求全部在队列里等待用户体感就是转圈圈转半天。更可怕的是长尾效应——AI模型的响应时间方差极大P99可能到了15秒这些慢请求会长期霸占线程把整个服务的吞吐拖垮。我自己踩过的坑早期做一个问答系统后端直接同步调外部大模型接口压测的时候40并发就把服务打崩了。jstack一看两百多个Tomcat线程全部BLOCKED在HTTP连接上一个都不剩。那时候才意识到同步阻塞模型在AI场景下不是性能问题而是架构问题。1.2 异步化的本质把等待变成可让出的资源异步化的本质其实很朴素线程不应该傻等IO完成。传统的同步模型里线程发起一个HTTP调用后自己进入阻塞状态直到响应返回才继续执行。这段时间线程什么事都没干纯粹是占着内存等网络包。异步化的思路就是发起调用后线程立即被释放回池子里干别的活等数据到达了再通过回调或者挂起恢复来处理。在AI场景这个特性尤其关键。因为AI接口调用中有大量的网络等待、模型排队、token生成耗时实际CPU计算量反而很低。举个例子一次单轮对话请求后端调LLM花了8秒其中纯CPU计算可能只有几百毫秒剩下7秒多是等网络、等对方的GPU排队。如果你用同步模型8秒内这个线程全程被占用。如果用异步化或者虚拟线程模型这8秒内线程资源可以被几十个其他任务复用系统的整体吞吐能力自然就上去了。但这里要澄清一个误区异步化并不会降低单个请求的RT它优化的是单位时间内能处理的请求数量也就是吞吐。用户该等3秒还是等3秒但同样硬件条件下系统能同时服务更多人。理解了这一层再去看各种异步化工具思路就清晰多了。2. Java异步化三件套怎么配合线程池、CompletableFuture、虚拟线程2.1 线程池把等待时间变成资源预算Java里最经典的异步手段就是线程池。AI应用里线程池大小不能像传统Web那样拍脑袋定它有明确的公式线程数 目标QPS × 单请求RT秒假设你的业务目标是支撑50 QPS接口平均RT为5秒那么需要的同时在途线程数就是50 × 5 250。这250个线程不应该放在Tomcat容器线程里而应该放在独立的业务IO线程池中。为什么因为容器线程还要负责接收HTTP连接、做参数校验、返回响应等基础工作如果被AI长任务占满连健康检查接口都响应不了。实际配置时我通常这样定义一个专用线程池ThreadPoolExecutor aiExecutor new ThreadPoolExecutor( 128, // 核心线程数按目标QPS × RT估算 256, // 最大线程数留出峰值余量 60L, TimeUnit.SECONDS, new ArrayBlockingQueue(500), // 有界队列防止内存被耗尽 new ThreadFactoryBuilder().setNameFormat(ai-invoke-%d).build(), new ThreadPoolExecutor.AbortPolicy() // 队列满了直接拒绝快速失败 );两个关键点队列必须用有界队列否则请求无限堆积会吃掉所有堆内存拒绝策略用AbortPolicy让过载请求快速失败返回错误码而不是在队列里熬到超时。线程名一定要用有意义的名称排查线上问题时jstack一眼就能看出哪些线程在忙什么。线程数的确定不是一劳永逸的要根据压测结果持续调整。我见过有人把线程池调到4096结果下游AI接口每秒最多处理20个请求大量线程全部卡在等待响应的路上内存占用飙升毫无意义。线程池的大小最终由下游瓶颈决定不是越大越好。2.2 CompletableFuture编排并发分支别把回调写成回调地狱线程池解决了谁来干活的问题但AI应用里还有活怎么编排的问题。一次用户提问后端可能要同时做几件事判断缓存是否有命中从向量数据库检索相关文档从搜索引擎或ES做关键词召回组装Prompt后调用LLM这些分支有的可以并行有的要合并结果后再继续。用传统Future加回调写出来的代码很快会陷入回调地狱。CompletableFuture是Java 8引入的异步编排利器它把异步任务抽象成可以组合的流水线用起来比手写线程同步舒服得多。CompletableFutureListDocument vectorFuture CompletableFuture .supplyAsync(() - vectorStore.search(question, 5), retrievalExecutor); CompletableFutureListDocument keywordFuture CompletableFuture .supplyAsync(() - searchEngine.search(question, 5), retrievalExecutor); CompletableFutureListDocument docsFuture vectorFuture .thenCombine(keywordFuture, this::mergeAndDeduplicate) .orTimeout(3, TimeUnit.SECONDS) // 检索最多等3秒 .exceptionally(ex - List.of()); // 检索失败降级为空列表这一步把两路检索并行跑最慢的一路决定了整体等待时间而不是串行累加。thenCombine把两个结果合并成一路orTimeout给整个检索过程设了上限exceptionally保证即使检索全挂了也不会让整个请求抛异常而是按空文档继续走。这个类有三个坑必须记牢。第一个坑默认使用ForkJoinPool.commonPool。如果你写的supplyAsync没有传第二个参数任务会提交到JVM全局共享的ForkJoinPool它的线程数默认是CPU核心数-1。AI场景下大量任务都在阻塞等待网络IO这点线程根本不够用一旦占满所有异步任务全部排队。任何耗时调用都必须显式传入自定义线程池回调里的子任务也一样。第二个坑异常容易丢失。CompletableFuture链路上任何一个环节抛异常如果后续没有exceptionally或handle兜底异常会被吞掉表现为请求莫名其妙超时日志里啥也没有。所有异步链路末尾必须挂兜底异常处理。第三个坑回调线程不可控。thenApply默认在触发它的那个线程执行thenApplyAsync才会重新提交到线程池。如果你依赖某个回调一定在哪个线程跑来关联上下文比如传递TraceId记得用thenApplyAsync指定线程池。2.3 虚拟线程JDK 21带来的伪同步写法如果说CompletableFuture解决的是编排问题那虚拟线程解决的是写代码的体感问题。JDK 21正式发布了虚拟线程它为大量阻塞任务而生。虚拟线程由JVM调度占用内存极小默认栈只有十几KB而平台线程默认1MB左右创建成本几乎可以忽略不计。当一个虚拟线程执行阻塞IO时它会自动释放底下承载它的平台线程载体线程让载体线程去跑别的虚报线程。这意味着你可以用同步阻塞的代码风格享受到异步化吞吐的好处。不用CompletableFuture不用回调直接写成// 定义一个虚拟线程池随用随建用完就扔 ExecutorService aiExecutor Executors.newVirtualThreadPerTaskExecutor(); // 业务代码里直接用同步风格 public String chat(String question) throws Exception { FutureString llmFuture aiExecutor.submit(() - llmClient.call(question)); FutureListDocument docFuture aiExecutor.submit(() - vectorStore.search(question)); String answer llmFuture.get(10, TimeUnit.SECONDS); ListDocument docs docFuture.get(3, TimeUnit.SECONDS); return buildResponse(answer, docs); }aiExecutor.submit配合Future.get代码看着像同步但每个阻塞调用都不占平台线程几十万个虚拟线程并发也是常态。这就是JDK 21带来的最大改变你不需要刻意学习响应式编程也能写出高吞吐的IO密集应用。但虚拟线程有三个禁区要避开。第一CPU密集型任务不要用虚拟线程它不会让CPU变快反而增加调度开销。第二别在synchronized块里做阻塞IOsynchronized是平台线程级别的锁虚拟线程一旦在锁内阻塞会钉住载体线程。第三避免使用JNI方法同样会钉住载体线程。AI应用里大量的是HTTP调用、向量检索这类IO操作放心用虚拟线程没问题。实际生产里我不会只用一种方案。通常的做法是用CompletableFuture做多路分支的编排用虚拟线程池做实际阻塞IO的落地用平台线程池或Tomcat容器线程接收请求和快速响应。各取所长。3. 高并发AI应用的拦截与保护层设计异步化把系统吞吐拉开之后新的问题来了你这边吞吐上去了下游AI接口能接住吗即使能接住你的内存和上游配额扛得住吗高并发不只是把请求发出去更重要的是发不出去的时候怎么办。这一层我把它叫保护层是整个高并发设计的重中之重。3.1 限流信号量和令牌桶到底该用哪个AI应用的限流有两条路控制同时在途的并发数和控制单位时间的请求速率。并发数控制最直接的工具是Semaphore。你的上游AI账号可能只允许5个并发请求那就在服务里声明一个Semaphore(5)每次要调AI接口前先tryAcquire()拿到了就调拿不到就排队或返回降级提示。这比请求进来后再被上游限流明智得多因为上游的一次429响应也是要花时间的。private final Semaphore llmConcurrencyLimiter new Semaphore(5); public String callLlm(String prompt) { if (!llmConcurrencyLimiter.tryAcquire(2, TimeUnit.SECONDS)) { throw new TooManyRequestsException(系统繁忙请稍后再试); } try { return llmClient.call(prompt); } finally { llmConcurrencyLimiter.release(); } }速率控制则用令牌桶思路。比如限制每秒最多进来20个请求用Guava RateLimiter或者自研一个简单的令牌桶都行。AI场景比较特殊我建议两种都做入口对全局限速防止流量洪峰打垮服务调用上游前对单账号并发限流防止被上游封掉账号。令牌桶和信号量的选型标准我总结成三句话如果你关心同一时刻最多多少人调用用信号量如果你关心每秒最多处理多少请求用令牌桶如果两者都关心先用令牌桶挡流量再用信号量保上游。3.2 超时、熔断与重试防连环雪崩的三道闸门高并发系统最怕的是一个慢请求拖垮全链路。AI接口一旦上游波动响应时间从2秒飙到60秒你这边如果无限期等待线程池很快就被这些慢请求占满随后所有新请求全部遭殃。所以三道闸门必须全部做。第一道闸门是超时。每一条AI调用都要设置连接超时、读取超时整个链路还要设置总预算。我用HTTP客户端举例HttpClient client HttpClient.newBuilder() .connectTimeout(Duration.ofMillis(800)) // 连不上赶紧放弃 .executor(aiExecutor) // 使用虚拟线程池 .build(); HttpRequest request HttpRequest.newBuilder(uri) .timeout(Duration.ofSeconds(15)) // 整个请求最多15秒 .POST(...) .build();注意timeout这个参数是整个请求从开始到读完响应的总时长但流式输出场景里只要持续拿到数据就不应该中断这需要按数据块来中断判断不能拿总超时一刀切。这里我用HttpClient的BodyHandlers.ofLines()配合Flow API逐行处理每行之间的间隔超过阈值就视为断流。第二道闸门是熔断。熔断的意思不是超时了重试一次而是连续失败多了直接进入拒绝状态不再发请求给上游。像Resilience4j这类库可以直接集成但AI场景我也见过手写熔断器的核心参数就是三个窗口大小统计最近N次请求、失败率阈值比如超过50%就熔断、恢复时间熔断后过多久放一个试探请求。AI模型上游不稳定是常态熔断能保护你的线程池和余额。// 一个极简熔断器统计最近20次调用 class LlmCircuitBreaker { private final int windowSize 20; private final double failureThreshold 0.5; private final AtomicLong windowStart new AtomicLong(); private final AtomicInteger requestCount new AtomicInteger(); private final AtomicInteger failureCount new AtomicInteger(); private volatile boolean open false; public boolean canPass() { if (open) { // 熔断状态下每5秒放一个试探请求 if (System.currentTimeMillis() - windowStart.get() 5000) { return true; } return false; } return true; } public void recordResult(boolean success) { ... } }第三道闸门是重试。这个在AI场景要额外的谨慎。大模型接口不是幂等的你重试一次它重新生成一次账单一分不少地扣。重试策略必须满足三个条件请求确实是网络中断或超时不是业务失败你有请求ID可以传给上游做去重重试间隔是指数退避1秒、2秒、4秒、最大10秒并且总的额外等待时间不超过你的链路预算。我的实践原则是宁可返回暂时不可用也不要为了成功率高而乱重试。AI服务的高价值请求可以用一次重试兜底普通对话请求重试一次就够了。3.3 背压与内存保护别让队列吃掉你的堆内存异步化系统有一个隐蔽的杀手——任务堆积。当请求进来的速度大于处理速度时所有未处理的任务都会堆积在队列里。如果队列是无界的最终结果就是把堆内存填满触发Full GC然后整个应用卡死。有界队列是第一步前面线程池示例里我用ArrayBlockingQueue(500)就是干这个的。第二步是入口信号量在请求进入业务层之前就做一次并发拦截防止瞬间洪峰把所有队列都塞满private final Semaphore requestLimiter new Semaphore(100); public void handleRequest(HttpServletRequest request, HttpServletResponse response) { if (!requestLimiter.tryAcquire()) { // 直接返回繁忙不进入后续处理 response.setStatus(503); return; } try { processAsync(request, response); } finally { requestLimiter.release(); } }第三步是考虑AI应用特有的内存压力大模型的输入输出token是有长度的。你在入口不做长度限制一个用户提交了5万字的问题你把全文塞进Prompt然后转发给模型这个请求占的内存、网络带宽、上游计费都是普通请求的几十倍。高并发场景必须对Prompt长度做限制超长的要么截断要么拒绝。另外响应结果的缓存比传统接口更值得做。同一个热门问题完全可以让后续用户直接命中缓存不再重复调用大模型。我见过一个客服问答系统加了Embedding相似度缓存之后上游调用量直接降了60%。缓存的是向量检索结果LLM回复用相似度阈值判断是否命中。4. 实操一个可复现的AI流式问答Agent骨架理论讲再多不如直接看一个能跑起来的骨架。下面这个示例我尽量保持代码简洁但保留了完整的高并发设计要点异步编排、虚拟线程、信号量限流、SSE流式输出、断连取消。4.1 整体架构与并发模型设计先描述一下目标场景Web页面提供一个对话框用户输入问题后后端需要做两路检索向量库ES再把检索结果拼进Prompt最后以流式方式把大模型的回复逐步推送到浏览器。要求是首字到达尽量快用户不用等所有token生成完才看到内容高并发下系统不死过载时快速失败客户端中途关闭页面后端要立刻停止调用大模型省钱并发模型设计如下Tomcat容器线程只负责接收HTTP请求、创建SSE连接、把任务丢给业务线程池后立即返回检索和LLM调用都在虚拟线程池里执行用CompletableFuture做两路检索的合并编排用有界队列信号量保护系统不被打垮。这里用SseEmitter它是Spring MVC原生支持的服务端推送对象用法简单兼容性好。配合虚拟线程池代码风格几乎是同步的写起来非常轻松。4.2 Controller层快速返回不做重活RestController RequestMapping(/api) public class ChatController { private final ChatService chatService; GetMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter chat(RequestParam(question) String question) { SseEmitter emitter new SseEmitter(60_000L); // 单次流式最长60秒 chatService.chatAsync(question, emitter); return emitter; } }注意Controller方法本身不做任何耗时操作它创建了SseEmitter然后立即调用chatAsync把工作异步化。这里的chatAsync内部会立刻启动新的虚拟线程去干活方法本身立刻返回Tomcat线程马上空出来接下一个请求。为什么要返回SseEmitter而不是直接等待结果呢因为大模型回答是流式的你可能在4秒后就要推送第一个token但完整的回答需要15秒。SseEmitter让你可以持续地向客户端推送数据而不是等全部完成再一次性返回。4.3 Service层异步编排主流程Service public class ChatService { private final VectorStore vectorStore; private final SearchEngine searchEngine; private final LlmClient llmClient; // 虚拟线程池专门跑阻塞IO任务 private final ExecutorService ioExecutor Executors.newVirtualThreadPerTaskExecutor(); // 检索结果合并线程池也可以复用上面的 private final ExecutorService mergeExecutor Executors.newVirtualThreadPerTaskExecutor(); // 限制同时调用大模型的并发数账号配额是5并发 private final Semaphore llmConcurrencyLimiter new Semaphore(5); // 入口限流防止服务过载 private final Semaphore entranceLimiter new Semaphore(100); /** * 异步处理聊天请求结果通过 SseEmitter 持续推送 */ public void chatAsync(String question, SseEmitter emitter) { if (!entranceLimiter.tryAcquire()) { emitter.completeWithError(new TooManyRequestsException()); return; } // 在虚拟线程池里跑整个流程Controller 立即返回 ioExecutor.submit(() - processChat(question, emitter)); entranceLimiter.release(); } private void processChat(String question, SseEmitter emitter) { try { // 1. 两路检索并行发出去 CompletableFutureListString vectorFuture CompletableFuture.supplyAsync( () - vectorStore.search(question, 5), ioExecutor); CompletableFutureListString keywordFuture CompletableFuture.supplyAsync( () - searchEngine.search(question, 5), ioExecutor); // 2. 合并检索结果最多等3秒失败降级为空列表 CompletableFutureString promptFuture vectorFuture .thenCombine(keywordFuture, (docs, keywords) - buildPrompt(question, concat(docs, keywords))) .orTimeout(3, TimeUnit.SECONDS) .exceptionally(ex - { log.warn(retrieve failed, fallback to direct chat: {}, ex.toString()); return buildPrompt(question, ); }); // 3. 拿到Prompt后流式调大模型 promptFuture.thenAcceptAsync(prompt - { streamLlmAndSend(emitter, prompt); }, ioExecutor); } catch (Exception e) { log.error(chat failed, e); emitter.completeWithError(e); } } }这个流程里有几个值得琢磨的设计点。第一thenAcceptAsync执行的时候已经拿到了完整Prompt它内部会再次在虚拟线程池里发起大模型流式调用。第二为了避免用户在检索阶段干等理想情况下应该在大模型能接受的条件下尽早把首字推出来实际项目里甚至可以把检索结果也做成流式推给前端展示。第三entranceLimiter用来挡住过载流量llmConcurrencyLimiter号在streamLlmAndSend内部再获取两者负责不同角色的限流。这段代码没有用Async注解因为Async底层是代理线程池配置起来绕来绕去不如直接显式声明ExecutorService一目了然。排查问题时能直接看到线程名知道任务跑在哪个池子里。4.4 流式输出与断连取消最容易被忽略的环节大模型流式调用的典型实现是用HTTP客户端读流拿到每个增量token就通过SseEmitter发出去。伪代码如下private void streamLlmAndSend(SseEmitter emitter, String prompt) { if (!llmConcurrencyLimiter.tryAcquire()) { emitter.send(SseEmitter.event().data(当前请求过多请稍后再试)); emitter.complete(); return; } try { HttpClient client HttpClient.newBuilder() .connectTimeout(Duration.ofMillis(800)) .executor(ioExecutor) // 重要用我们的虚拟线程池 .build(); HttpRequest request HttpRequest.newBuilder() .uri(URI.create(https://api.example.com/v1/chat)) .timeout(Duration.ofSeconds(30)) .header(Authorization, Bearer xxx) .header(Accept, text/event-stream) .POST(BodyPublishers.ofString({\prompt\:\ escapeJson(prompt) \,\stream\:true})) .build(); // 注册SseEmitter的断开回调一旦客户端断开就取消请求 AtomicBoolean cancelled new AtomicBoolean(false); emitter.onCompletion(() - cancelled.set(true)); emitter.onTimeout(() - cancelled.set(true)); client.sendAsync(request, BodyHandlers.ofLines()) .thenAcceptAsync(response - { response.body().forEach(line - { if (!cancelled.get()) { try { String token parseSseToken(line); if (token ! null) { emitter.send(SseEmitter.event().data(token)); } } catch (Exception e) { // 客户端断开会抛 IOException这里必须终止 cancelled.set(true); return; } } if (cancelled.get()) { // 让上游断开流 throw new CompletionException(new IOException(client cancelled)); } }); }, ioExecutor) .exceptionally(ex - { log.warn(stream interrupted or failed: {}, ex.getMessage()); return null; }) .thenRun(() - { llmConcurrencyLimiter.release(); emitter.complete(); }); } catch (Exception e) { llmConcurrencyLimiter.release(); emitter.completeWithError(e); } }这段代码有几个细节很关键。onCompletion和onTimeout回调里不能直接阻塞式地关闭HTTP连接所以我用AtomicBoolean做标志在下一次处理token时检查。emitter.send如果抛异常说明客户端已经断开这时候要把整个流处理链路终止掉否则大模型还在继续生成token每一秒都在烧钱。注意客户端断开后必须调用远端接口的cancel或关闭响应流。有些HTTP客户端库不会自动中断你要在onCompletion回调里显式地response.cancel()。省下来的每一个token都是真金白银。4.5 参数计算实例50 QPS的配置推导光给代码不给参数不算真正可复现。下面我给一个具体场景的完整配置推导过程。假设业务目标峰值50 QPSAI接口平均RT 4秒两路检索平均RT 1秒上游AI账号限制5并发。第一项虚拟线程池大小。50 QPS × 4秒 200个同时在途的AI请求。虚拟线程本身很轻量但也不能完全无脑开我设置newVirtualThreadPerTaskExecutorJVM按需创建只要底层载体线程够用即可。载体线程默认数量与CPU核数相关实际IO等待时大部分会被释放所以提供200个并发虚拟线程没有压力。但注意HTTP连接池、下游连接数都是物理上限虚拟线程再多也不会超过这些瓶颈。第二项检索线程池。50 QPS × 1秒 50个并发检索任务。复用ioExecutor即可注意虚拟线程池创建任务无上限这里真正限制并发的是下游检索服务的连接数所以检索客户端的连接池要能扛住50并发。第三项信号量参数。llmConcurrencyLimiter new Semaphore(5)这是硬限制来自上游账号配额。这里会出现一个现象虚拟线程池允许200个任务但信号量只放5个进AI调用其余195个虚拟线程会阻塞在tryAcquire上。这是完全正常的虚拟线程挂在信号量上内存占比小系统不会崩只是请求排队。第四项入口信号量。目标是快速失败我设置100。超过100个同时在途请求直接返回503用户看到的提示是系统繁忙而不是无限等待。第五项队列与拒绝策略。如果不用信号量而是用线程池队列我建议队列长度设为QPS × 容忍的排队秒数。假设容忍排队2秒队列长度就是50 × 2 100。超过之后AbortPolicy触发快速失败。5. 常见问题与排查技巧实录5.1 线程池被打满从哪里下手查典型症状请求大量超时服务整体吞吐掉到接近零。第一步先jstack看线程状态分布重点看两种大量线程WAITING (parking)说明线程池任务队列在排队或者CompletableFuture在等待结果大量线程RUNNABLE且堆栈停在SocketInputStream.read说明线程都卡在等待网络IO用socketTimeout和连接池上限去排查看线程名很关键这就是前面强调必须自定义线程名的原因。如果线程名都是ai-invoke-%d那就去查这个池子的任务是谁提交的、下游为什么慢。5.2 CompletableFuture默认线程池饥饿有段时间我做RAG检索supplyAsync没传线程池压测30并发时莫名大量超时日志里却没有任何异常。后来用jstack发现commonPool线程全部WAITING在检索调用上但CPU核心数只有8commonPool最多7个线程全被阻塞IO占住后续任务全部排队。这个坑非常隐蔽因为代码看起来完全正常只是超时。检查方法很简单在所有supplyAsync、thenApplyAsync、runAsync调用上加断点看是否有调用没有传Executor参数。或者全局搜一下Async方法逐个确认。这个错误在我团队里至少被踩过三次。5.3 虚拟线程下连接池反而成了新瓶颈虚拟线程可以创建成千上万个但你的数据库连接池、Redis连接池、HTTP连接池还是物理限制。最常见的错误是虚拟线程开了5000个全部去抢数据库连接的信号量结果连接池只有50个4950个虚拟线程全堵在等连接上数据库负载没涨应用倒是先卡死了。解决办法是连接池大小的确定要放在虚拟线程数之前。先算下游并发上限再决定你的虚拟线程规模。比如数据库最大连接数是50那你全局的虚拟线程并发度就别超过几百否则徒增无意义的线程切换和排队。5.4 客户端断开后上游还在持续计费流式场景下用户不会每次都很温柔地等token全部返回。关网页、切应用、刷新页面都意味着SSE连接断开。如果后端没有在onCompletion里取消上游请求模型服务会一直生成到自然结束。我见过真实的账单惨案一个几十人的内测系统一个下午因为断连未取消花了平时半个月的模型调用费用。建议在代码里显式把断连纳入主流程处理而不是当作异常边缘情况。SseEmitter的onCompletion回调里不但要置取消标志最好还能拿到底层HTTP连接的引用直接关闭。5.5 涉及面试官常问的几个点怎么答这个话题经常出现在Java面试题讨论里我踩过的坑基本就是面试官想挖的点。核心问题是AI应用的高并发设计和传统Web高并发有什么异同好的回答思路是先点明负载模型差异长耗时、外部不可控、流式再说线程模型差异同步阻塞不可行然后落到工具选型CompletableFuture编排虚拟线程落地最后提到保护层限流、熔断、超时、背压。面试官会追问CompletableFuture默认线程池有什么坑虚拟线程和平台线程的区别AI接口该不该重试都是一线问题把我上面写的经验答出来基本就够了。6. 最后再分享一点实操体会文章写到这里我把自己在Java AI应用异步化与高并发设计上的沉淀做了个完整梳理但每个系统都有自己特殊的地方。落到自己的项目时请第一个去看你的下游瓶颈在哪里是模型接口并发限制、是向量库性能、还是网络带宽。先确定瓶颈再去设计线程池、信号量和队列顺序反了容易白忙活。我的体会是Java 21之后做AI应用后端技术上的标准答案越来越清晰了虚拟线程负责解决IO密集的吞吐问题CompletableFuture负责解决多路调用的编排问题限流熔断超时负责解决外部依赖不稳的问题。三者各管一段不需要引入复杂的流式编程框架也不需要靠调整堆大小硬扛。最后一个小建议做这种系统的压测时不要只看总QPS。把P50、P95、P99分开看重点观察线程池活跃度、信号量排队数、下游接口拒绝次数这三个指标。只有它们同时处在一个合理范围你的异步化架构才算真正立住了。如果发现P99涨得飞快而P50还好多半是排队问题——这时候不是异步化错了是入口限流或队列容量还需要再调整。