做 AI 应用最难受的时刻之一,就是你明明调的是流式接口,结果用户那边看到的还是一个转圈圈,过了四五秒才“哗啦”一下全出出来。更难受的是,后端根本没报错,线程池却悄悄死了,接口一个接一个超时,进程没挂,但整个服务已经“假死”。这种问题我见过太多次,也亲手排查过太多回,根子往往就一句话:线程被堵死了。这篇文章围绕“AI 流式输出”和“异步调度”这两件事展开,核心始终是标题里那五个字——别堵死线程。我会把流式接口的底层协议、线程池被堵死的完整连锁过程、以及从同步到异步的改造实操都讲透,包括很多不踩坑你根本不会留意的细节。不管你做的是 Java 还是 Python 后端,是业务接口还是网关层,都能直接拿去参考。1. 流式输出的本质:一次“打字机”式的技术改造1.1 从“等结果”到“看结果”的体验飞跃先聊为什么大家非要流式。用过 ChatGPT、Claude 这类产品的人都有感觉:AI 生成回答不是瞬间完成的,一个三四百字的回答,模型可能要“想”5 到 20 秒。如果走传统的同步接口,这 5 到 20 秒里用户只能盯着一张空白页面,什么反馈都没有。产品数据也说明问题:超过 3 秒没有反馈,用户流失率会直线上升。我自己做过的 AI 问答项目里,首字延迟从 8 秒压到 1 秒以内,用户满意度提升非常明显,这就是流式的价值。流式输出(Streaming Output)解决的就是这个体验问题。它把模型一次性的长响应,拆成一个个 token(你可以理解为词或子词),边生成边推送。用户在界面上看到的效果是文字一个字一个字往外蹦,像有人在打字。很多 AI 产品的“打字机效果”就是这么来的。但这里有个容易忽略的坑:流式只是改变了“数据怎么返回”,并没有改变“模型算多久”。从调用方视角看,总耗时是一样的,甚至因为要持续维持连接,占用的资源更多。也就是说,流式把压力从“用户等待”转移到了“后端长连接维护”上,这才是后面一堆线程问题的根源。很多人以为用了流式就万事大吉,结果发现服务端更吃紧了,原因就在这里。1.2 流式接口底层到底发生了什么拿目前用得最多的 OpenAI 兼容接口来说,流式响应走的是 SSE(Server-Sent Events)。SSE 本质就是 HTTP 长连接,服务端按text/event-stream格式,一条一条往客户端推数据。每条事件以data:开头,以空行结束。AI 流式接口的事件体通常是 JSON,里面带choices[0].delta.content字段,收到的就是一个增量片段。举个例子,真实抓包看到的流大概是这种:data: {choices:[{delta:{content:你},index:0}]} data: {choices:[{delta:{content:好},index:0}]} data: {choices:[{delta:{content:,},index:0}]} ... data: [DONE]注意几个细节:第一,这串内容极其碎片化,一个词可能拆成好几次推送,甚至一个字分多个事件;第二,HTTP 层用的是 chunked transfer encoding,也就是 TCP 连接上数据是一块一块流过来的,不是一次性写完;第三,如果客户端没及时读数据,服务端缓冲区满了会反压(TCP backpressure),这时候模型那边也会等。这就带出一个关键认知:fetch、HttpClient、requests 这些库都能处理流式响应,关键看你怎么调。大多数 HTTP 客户端默认“等全部响应体到达再返回”,即使你设置了streamtrue,底层如果没配合逐行读取,一样会攒到全部数据才 yield。我见过不少“明明用了流式接口,前端还是等全量出来”的案例,问题就出在这一层。所以别以为配个 stream 参数就完事了,读响应的方式才是决定成败的细节。2. 为什么你的线程会被“堵死”2.1 同步等待是万恶之源在线程模型下,一次同步调用意味着“当前线程发起请求后,进入等待状态,直到响应完全返回”。AI 接口的响应时间有多长?慢的时候 60 秒甚至更久。这段时间里,那条线程做不了任何事,就在那等着。你可以把线程想象成餐厅的服务员,如果每个服务员都站在一张桌子旁边等客人慢慢吃完,这餐厅就没法接待别的客人了。假设你的服务用的是 Tomcat 默认配置,工作线程上限 200。某个活动上线后,10 个用户同时发起 AI 对话请求,每个请求都要等 20 秒。这 10 条线程就被占住了,相当于 10/200 的容量没了。用户一多,比如 200 个并发请求,所有工作线程全部被占满,后面的请求直接排队,排不上就超时。你在监控里看到的往往就是“TPS 掉到 0 但 CPU 只有 5%”,因为线程全部在等网络,根本没在算。Java 里这种问题最容易藏。很多同学用 RestTemplate、OkHttp 同步调用大模型接口,一个execute()下去,当前线程就进去了。代码看起来只有一行,非常清爽,但坑就在下面深着呢。用Thread.currentThread().getName()把执行线程名打出来看看,你会发现同样一条 Tomcat 工作线程,从请求进来一直陪到 AI 返回,中间啥也没干。2.2 线程池耗尽:一次连锁反应的现场还原我复盘过好几次线上假死,套路都差不多。这里还原一遍,方便你对号入座。第一步,业务高峰期,多个请求同时调 AI 流式接口,每个占住一条 Tomcat 工作线程,等待模型返回。第二步,Tomcat 的 200 条线程被占满,后续所有请求(包括不涉及 AI 的普通接口)全部排队。第三步,等待队列塞满,新请求触发拒绝策略,默认的 AbortPolicy 直接抛RejectedExecutionException,调用方 500。第四步,前端收到 500 就自动重试,重试又涌进来,线程池进一步被冲垮。更糟的是,如果这些请求是从某个 Spring 异步线程池发出去的,情况会复杂一层:业务线程池也在等,HTTP 连接池也在等,连接池超时后又波及到别的服务。这就是典型的“一线程堵死,全线崩溃”。问题不在 AI,而在你把一条宝贵的线程押在了不可控的等待时长上。线程池这东西,看起来只是“线程的集合”,实际上它是整个系统的排队机制,一旦它被打满,所有依赖它的调用都会跟着遭殃。2.3 线程池参数怎么配:算给你看如果一定要用线程池(有些场景确实只能用线程池),参数不能拍脑袋。核心公式是:对于 IO 密集型任务:线程数 CPU 核数 × (1 等待时间 / 计算时间)这里的“计算时间”对 AI 流式场景来说,基本就是 JSON 解析和业务处理的耗时,通常在几毫秒到几十毫秒;而“等待时间”是模型响应时间,15 秒起步。假设 1 次请求等待 15 秒,计算 10 毫秒,比值是 1500。一个 8 核机器按公式算出来理论线程数是 8 × (1 1500) 12008 条线程。这个数字显然不能直接照搬,因为线程是又贵又不可能无限开的资源,每条线程默认 1MB 栈空间,12000 条线程光栈就 12GB,内存直接爆。所以实战中要反过来算:先看你能给多少内存,再定线程数上限。8GB 堆、单线程默认栈 1MB 的设备,线程数压到 500 以内比较稳妥。再叠加下游 AI 服务的并发限制(比如每分钟 60 次)和 HTTP 连接池上限,真正合理的线程数往往是 100 到 300 之间。配置线程池的顺序应该是先算内存上限,再算下游限流,最后才是 CPU 公式。这个顺序很多人反着来,导致各种诡异问题。3. 异步调度的正确打开方式3.1 Java 的异步进化:从 Future 到 CompletableFuture堵死线程的解法不是不创建线程,而是别让线程“干等”。Java 里最基础的异步是Future,允许你提交任务后不阻塞,后面再get()。但Future.get()本身就是同步阻塞,如果你在业务代码里无脑调用,问题照旧。真正的异步模型应该是:提交任务、立即返回、回调通知,全程不等待。正确姿势是用CompletableFuture。它支持回调、组合,最关键的是supplyAsync()可以指定独立的线程池,把 AI 调用从 Tomcat 请求线程挪到专门的工作线程上,请求线程立刻释放。比如:CompletableFutureString future CompletableFuture.supplyAsync( () - aiClient.complete(prompt), // 这个阶段在 aiExecutor 中执行 aiExecutor ); future.thenAccept(result - log.info(AI 返回完成,当前线程:{}, Thread.currentThread().getName()));重点是thenAccept之后的逻辑不会占用 Tomcat 工作线程。future本身可以被继续传递,也可以配合CompletableFuture.allOf()等所有 AI 子任务都完成后统一处理。如果你要“等全部完成再汇总”,别用循环future.get(),那是变相把异步变回同步;应该用 allOf:ListCompletableFutureString futures prompts.stream() .map(p - CompletableFuture.supplyAsync(() - aiClient.complete(p), aiExecutor)) .collect(Collectors.toList()); CompletableFutureVoid allDone CompletableFuture.allOf( futures.toArray(new CompletableFuture[0])); allDone.join(); // 这一行只在自己确实需要同步等待时才用这里有个非常容易被忽略的点:如果回调也在aiExecutor里执行,回调里的代码也会有同样的问题。所以我习惯给不同阶段分开线程池,或者干脆用事件驱动的方式,不让回调做重计算。另外注意,CompletableFuture默认用的是 ForkJoinPool,里面的线程数等于 CPU 核数,你如果把 AI 这种几十秒的 IO 等待任务往里塞,这个池子瞬间被占满,项目里其它所有用 CompletableFuture 的地方都会跟着卡住。给 AI 调用单独建线程池是必须的,不是可选项。3.2 Python 那边的玩法:协程与线程池的搭配Python 走的是另一条路:GIL 决定了纯线程在 CPU 密集型任务上没法并行,但对 IO 等待反而友好。因为等待期间线程让出 GIL,别的线程能跑。问题是,如果你用requests这种同步库,即使底层是 IO 等待,线程也会被占住,只是 GIL 让出了 CPU 而已。从线程占用角度看,占住就是占住,和 GIL 没关系。Python 的推荐做法是异步框架加异步 HTTP 客户端。FastAPI httpx 的AsyncClient.stream()是个典型组合,流式读取 SSE 时不会阻塞事件循环:from fastapi import FastAPI from fastapi.responses import StreamingResponse import httpx app FastAPI() async def sse_proxy(prompt: str): payload { model: gpt-4o-mini, messages: [{role: user, content: prompt}], stream: True } async with httpx.AsyncClient(timeouthttpx.Timeout(60.0, read60.0)) as client: async with client.stream( POST, https://api.example.com/v1/chat/completions, jsonpayload ) as resp: async for line in resp.aiter_lines(): if line.startswith(data:): yield line \n\n app.get(/chat) async def chat(prompt: str): return StreamingResponse(sse_proxy(prompt), media_typetext/event-stream)注意timeout有多个参数:connect 超时和 read 超时是两回事。AI 流式接口的 read 超时不能设太短,因为模型可能“想”很久不吐词,但你又不能完全没有超时。我的经验是把 connect 超时设在 3 到 5 秒,read 超时设在 60 到 120 秒,再把一个“首个 token 超时”的逻辑做在应用层(比如 30 秒内没收到第一个 data 就断掉)。Python 还有一个常见的坑:线程嵌套线程。有些同学为了“更并发”,在异步代码里偷偷调threading.Thread,或者用asyncio.to_thread包一层同步库,结果主事件循环没堵,但子线程全堵在下游接口上,线程数越开越多。Python 线程切换本身也有开销,特别是在threading混合asyncio的场景,很容易出现上下文切换暴涨。原则是:一条链路尽量只用一个并发模型,要么全异步,要么全线程池,别混着来。顺带提一句,Python 3.13 开始有了 free-threading 模式,可以真正并行跑线程,但对 IO 等待场景帮助不大,因为等待本来就会让出 GIL。不要因为听说 GIL 要没了,就把同步阻塞代码到处撒线程,线程创建和切换的开销依旧在那。3.3 响应式编程是不是唯一出路很多人一听异步就说要用 WebFlux、Reactor。响应式确实是解决线程阻塞的好办法,在 Spring 生态里,FluxString配合 SSE 可以直接把流式响应串成响应式流:GetMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxString chat(String prompt) { return Flux.create(sink - { aiClient.completeStream(prompt, sink::next, sink::complete); }).subscribeOn(Schedulers.boundedElastic()); }但响应式不是银弹。它最大的问题是调试困难,调用栈是断的,异常处理逻辑跟同步代码完全不同,团队没有响应式经验的话,上线后维护成本很高。我做过的项目里,真正的选择依据是:如果只是“代理大模型流式响应给前端”,那我更倾向于用最简单的方式——前端能直接连的就直接连,不能直接连的用 Spring 的SseEmitter或者 Servlet 3.1 的异步支持,别一上来就上 WebFlux。如果你在做一个高并发网关,要支撑成千上万的 SSE 长连接,那响应式才划算。技术选型不是越新越好,是越合适越好。Redis 的线程 IO 模型就很有启发性:主线程用单线程事件循环处理命令,网络 IO 由多线程处理,真正体现了“不等待”的哲学。做 AI 网关的时候,可以参考这种思路,把网络读写和业务处理解耦,而不是靠堆线程硬扛。4. 实操改造:一个完整的流式接口从堵到顺4.1 改造前后的代码对比纸上谈兵没意思。我拿一个真实的小项目改造过程来演示,原先是 Spring Boot RestTemplate 同步调 AI,改成 Spring MVC SseEmitter 独立线程池。这个项目是个简单的 AI 聊天问答服务,前端只有一句话输入框和一块显示区域,但上线后并发一高就频繁超时,这才被迫做了改造。改造前的问题代码大概长这样:GetMapping(/chat) public String chat(String prompt) { // 这一步会阻塞 Tomcat 工作线程最长 60 秒 ChatResponse resp restTemplate.postForObject( https://api.example.com/v1/chat/completions, buildRequest(prompt, false), ChatResponse.class ); return resp.getContent(); }这段代码在低并发下没问题,一压测就原形毕露。改造后:GetMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter chat(String prompt) { SseEmitter emitter new SseEmitter(120_000L); // 2 分钟超时 aiExecutor.execute(() - { try { aiClient.completeStream(prompt, chunk - { emitter.send(SseEmitter.event().data(chunk)); }); emitter.complete(); } catch (Exception e) { emitter.completeWithError(e); } }); return emitter; }关键改动只有两处:第一,接口不再返回字符串,而是返回SseEmitter,Spring 会拿它做 SSE 推送;第二,真正的 AI 调用丢给aiExecutor执行,Tomcat 请求线程立刻返回。Tomcat 工作线程的占用时间从“最长 60 秒”降到了“几毫秒”。这里有个细节:SseEmitter 超时时间一定要大于你的 read 超时。如果 AI 最慢 60 秒返回,你把 emitter 超时设成 30 秒,客户端就会莫名断连,看起来像 AI 吐了一半被砍了。同样,前端 fetch 也要把readTimeout调大,否则浏览器 20 秒没收到数据就报网络错误。这种问题最坑,因为后端日志显示一切正常,断连发生在浏览器那一层。4.2 异步网关层如何把流“接力”给前端如果你的架构里还有一层网关(比如自研的 API 网关,或者公司的 BFF 层),流式响应在每一层都得“边收边转”。最容易翻车的地方是网关层用同步 HTTP 客户端去请求后端,要么全量等完再往前端吐,要么直接超时。正确的接力逻辑是:网关收到后端传来的 SSE 流,逐条解析data:行,然后立即把同一段数据写到下游响应。也就是从“后端 → 网关 → 前端”这条链路上,每一跳都是流式的。如果用 Java 写网关,HttpURLConnection和旧 HttpClient 都不支持流式读响应体,建议直接用java.net.http.HttpClient(Java 11 起内置)的BodyHandlers.ofInputStream(),或者用 OkHttp 的onResponse回调里逐行读。另外,网关层最好别做“攒批”:每收到几个 chunk 才往客户端推一次,来降低推送频率。这种做法表面上看似省 IO,实际上会在链路上引入额外延迟,而且一旦中间断开,损失的已经不是一条消息,是一整批。我的实践是:来什么推什么,保持流数据的透传。压缩可以开,但要用流式压缩(如 gzip 的连续模式),不能等整个流结束才压缩,否则就失去了流式的意义。4.3 承载量评估:改造后能扛多少并发读者最关心的肯定是“改完到底能扛多少”。我拿一个真实压测数据来说明。某项目原先用同步方式,8 核 16G 的机器,Tomcat 默认 200 线程,压测 50 并发,每个请求平均耗时 20 秒,实际吞吐不到 10 QPS,线程池直接打满。改造后用 200 线程的独立aiExecutor处理 AI 调用,Tomcat 工作线程只负责接收和返回 emitter。压测结果:同一台机器,200 并发,每个请求体验不受影响,吞吐接近 150 QPS,而且 Tomcat 工作线程占用率不到 60%。瓶颈转移到下游 AI 服务的限流上。这个提升的核心原因是:Tomcat 工作线程不再闲置等待,线程复用率大幅提高。说白了,200 条线程在同步模型下顶多同时服务 200 个“正在等待中”的请求;在异步模型下,200 条线程可以服务几千个“已受理、等待中”的请求,因为每个请求真正占用线程的时间只有几毫秒。这就是“别堵死线程”的量化意义。5. 常见问题与排查技巧实录5.1 线程死锁、池耗尽、连接泄漏三巨头AI 流式场景里,线上问题逃不出这三类,我一个个说。线程死锁:典型的现场是 A 线程等待 B 线程的结果,而 B 线程也在等 A 线程的资源。在 AI 调用场景,死锁经常藏在“AI 超时重试”的代码里。比如线程池被打满后,重试任务又被提交到同一个已满线程池,新任务永远等不到执行,形成变相死锁。另一个常见情况是CompletableFuture的默认 ForkJoinPool 被 AI 阻塞任务占满,导致所有使用它的异步任务集体卡住。所以给 AI 调用单独建线程池是必须的,还要给get()和join()设置超时,不能让等待无限继续。线程池耗尽:这个上文分析过。排查时先看线程数。Linux 下top -H -p pid能看到线程快照,如果发现有几百条线程状态都是D(不可中断睡眠)或长时间S(睡眠),基本可以断定它们在等网络 IO。拿jstack导出线程转储,grep一下 WAITING 状态线程的调用栈,十有八九能看到指向 HTTP 客户端read方法的那一行。连接泄漏:很多 AI HTTP 客户端默认不会无限复用连接。如果你在每次请求里 new 一个 OkHttpClient,或者不关闭响应体,连接池会被慢慢耗尽。更隐蔽的是连接池的maxConnections设置太小,比如默认 20,而你开了 100 个并发流,剩下 80 个请求全部在等连接,表现起来和线程池耗尽一模一样。排查这种问题,用lsof -p pid | grep TCP看看连接数,再检查 HttpClient 的池参数。5.2 现场排查:拿到线程转储怎么读jstack 是 Java 排查线程问题的核心工具,但很多人拿到输出后一头雾水。我说一个快速的阅读方法。先看状态分布,统计 RUNNABLE、WAITING、BLOCKED、TIMED_WAITING 各自的数量。正常服务 RUNNABLE 应该占大头;如果 WAITING 占绝大多数,而且线程名集中在http-nio-8080-exec-*,那基本可以断定是工作线程池被打满。找到一条可疑线程,看它的调用栈底部。比如看到:at java.base/sun.nio.ch.NioSocketImpl.read(NioSocketImpl.java:139) at org.apache.hc.client5.http.impl.io.ChunkedInputStream.read(...)这说明线程卡在“读分块响应”上,也就是正在等上游 AI 吐数据。结合线程数统计,如果大量线程都停在这里,问题定位就完成了:下游 AI 响应慢,导致工作线程被占满。Windows 下有人会用 Process Explorer 看线程,甚至可以冻结(freeze)某个线程来模拟卡死现象。冻结线程对本地调试重启时的条件竞争很有用,但千万别在生产环境这么干,冻结的是进程里某个线程的调度,会让依赖它的请求直接假死,而且 Process Explorer 并不会告诉你为什么会卡,它只是给你一个“手动制造故障”的工具。线上排查还是靠线程转储和监控日志。5.3 排查工具与避坑清单最后整理一份实用的避坑清单,这些都是我踩过或者帮别人擦过屁股的:第一,所有 HTTP 超时都要显式设置。连接超时 3 到 5 秒,读取超时 60 到 120 秒,并且最终响应(整体)超时要比这些更大。第二,给 AI 调用单独配置线程池,不要在Async默认执行器、ForkJoinPool 或 Tomcat 工作线程里跑模型调用。第三,流式接口一定设置媒体类型为text/event-stream,并且关闭响应缓冲。Spring 里如果忘了设produces,或者 Nginx 层开了proxy_buffering on,前端可能等全量结束才看到数据,等于白做流式。第四,线程池要用有界队列和合理的拒绝策略。队列选ArrayBlockingQueue,别用无界LinkedBlockingQueue,否则任务无限排队,内存先爆。拒绝策略要么CallerRunsPolicy,要么自定义告警,不要让拒绝默默发生。第五,守护线程要慎用。如果你用ExecutorService创建了守护线程处理 AI 回调,当主线程结束、JVM 里只剩守护线程时,进程会直接退出,长连接全部断开。自定义线程工厂时把线程设为非守护。第六,AtomicInteger在计数线程池任务数、请求数时没问题,但别拿它来做线程安全的业务计数,因为它只保证单个变量的原子性,不保证复合操作的原子性。真要统计 A 线程池里的活跃任务数,还是老老实实等ThreadPoolExecutor的getActiveCount,或者自己加锁保护。第七,在 Python 异步代码里混用同步阻塞库,是线程切换开销的隐形杀手。每个asyncio.to_thread都会创建线程,如果循环调用,线程数会失控。能用异步库(httpx、aiohttp)就尽量别用同步库。第八,针对“线程切换时会泄漏吗”这个问题,答案是:切换本身不会泄漏,但线程局部变量(ThreadLocal)在线程复用池里会。池里的线程不会因为任务结束就被回收,ThreadLocal里的数据如果没有清理,下一个任务复用这条线程时能读到上一个任务的值。AI 请求里如果往 ThreadLocal 塞了用户上下文,记得在 finally 里 remove。第九,开始排查之前先确认你看到了真相:连接池耗尽、线程池耗尽、下游慢、上游超时,这四类问题的表象都是“请求变慢、开始超时”,但解法完全不同。拿线程转储、连接数、GC 日志三个数据源交叉验证,别靠猜。6. 写在最后的几句体会其实不存在“绝对正确”的架构,只要记住一点:在 AI 这种动辄几十秒的 IO 等待面前,任何让线程干等的设计都是犯罪。线程是拿来干活的,不是拿来排队的。你真正需要的是把“发起调用”和“等待结果”拆开,用事件或者回调把结果送回来,而不是让一条线程从头盯到尾。我个人实操中的体会是:一开始别追求什么高大上的响应式,先把同步调用改成异步线程池 回调,把超时和连接池配好,能解决 80% 的问题。剩下 20% 的高并发场景,再考虑 WebFlux 或者独立的 SSE 网关。技术的选择要跟着你的流量走,而不是跟着技术热度走。希望这篇文章能帮你少踩几个坑,下次线上再“假死”的时候,至少能知道该从哪里下手。