企业官网建设流程全解析

做AI推理服务的人应该都有过这种体验：一个识别请求，输入一张几MB的图片，模型跑了一两秒，然后一次性把结果全部返回。早期这样没问题，但现在大模型、多模态、视频推理的场景越来越多，一个响应动辄几百MB甚至几GB，推理耗时几十秒甚至几分钟，再走传统的"请求-响应"路线，第一个字出来之前用户已经等疯了。这类场景里，RPC传输层几乎必然要用到流式方案，而在gRPC Java生态里，这个方案的入口，就是StreamObserver。

这篇文章我打算把StreamObserver从原理到落地完整讲一遍：为什么AI场景需要它、它和普通RPC调用到底差在哪、怎么在一个真实的推理服务里把它用起来，以及我实际运维中踩过的那些报错。内容偏实践，适合正在把AI模型接进业务系统、被大响应体量和长耗时折磨过的后端工程师。看完你至少能明白：你的服务该不该上流式传输，上了之后哪些地方最容易翻车。

1. 为什么AI场景下RPC传输会变成瓶颈

1.1 传统RPC模型的三个"不适应"

传统的RPC调用，本质上是"一问一答"：客户端发一个请求，服务端吭哧吭哧算完，把完整结果打包回来，连接关闭。这个模型在普通业务接口上没什么问题，到了AI推理场景就完全变了，主要矛盾集中在三个地方。

第一是响应体量。AI的输出不是几个字段，而是可能包含原始张量、特征向量、合成音频、生成视频片段。一个向量检索接口返回几千条特征数据，轻轻松松几十MB；一个文生图服务把中间过程和最终图片一起返回，那就奔着上百MB去了。一次性把这么大的响应包在内存里组装完，再序列化传输，JVM堆直接报警。

第二是响应时间。深度模型的推理时间以秒甚至分钟计，传统RPC客户端发完请求就阻塞等待，中间完全拿不到任何进度信息。用户那边只能看到"转圈中"，体验非常糟糕。更现实的问题是，很多网关和负载均衡器有响应超时限制，一个请求挂在那里几分钟，中间链路早就把它掐断了。

第三是消费方式。AI推理的结果天然是增量的：大模型的token是一个一个生成出来的，语音识别是一段一段吐字的，视频推理是一帧一帧出结果的。用户希望能"边生成边看到"，而不是等全部生成完再一次性展示。传统的一元RPC（Unary RPC）根本没有办法表达"部分结果先到"这层语义。

1.2 流式传输恰好击中痛点

流式传输的思路很直白：响应不打包成一个整体，而是拆成多个消息，一有中间结果就立刻推给客户端。好处是立竿见影的：内存占用从"整个响应体量"降到"单个消息体量"；首包延迟从"全部推理完成"提前到"第一块结果生成"；客户端还能根据自己的处理能力决定读多快读慢。

这里我举一个最直白的例子。部署过LLM推理服务的人都知道，用户体验好坏最核心的指标是TTFT（Time To First Token，首token延迟）。如果用一个一元RPC接口去做聊天机器人，用户问完问题之后屏幕上至少要空白好几秒，等整个回复全部生成完才一次性刷出来，这体验基本等于没做。而用服务端流式RPC，模型每生成一个token就通过onNext推一次，用户在第一个token落地的瞬间就能看到文字在跳动，心理感知上快了不止一个量级。

1.3 StreamObserver在这里是什么角色

StreamObserver不是一套独立的框架，它就是gRPC Java里用来表达"流"的接口。你在proto文件里定义一个返回stream的方法，gRPC插件生成的服务端代码里，方法签名就会带一个StreamObserver参数；生成的客户端Stub里，方法签名也会要求你传入一个StreamObserver。所有流式交互，服务端往这个Observer上推数据，客户端从自己实现的Observer上收数据。

之前有个同事问我，流式RPC和WebSocket有什么区别，为什么不用WebSocket要费劲搞StreamObserver。这个问题其实问到点子上了。AI服务内部，模型服务之间往往是gRPC集群在通信，用StreamObserver意味着：第一，不需要额外维护一套WebSocket连接池和心跳；第二，gRPC天然支持多路复用，一条HTTP/2连接上同时跑几十个流，连接管理成本低得多；第三，流式RPC仍然保留了RPC的方法调用语义，拿到的还是结构化的消息体，而不是裸文本帧，业务代码写起来舒服得多。

2. StreamObserver的真实面目

2.1 接口结构与生命周期

先看这个接口长什么样。

public interface StreamObserver<V> { void onNext(V value); void onError(Throwable t); void onCompleted(); }

就这么简单，三个方法。但这个接口背后有一套非常严格的调用约定，踩坑的人十个里有八个都是因为没搞清楚这套约定。

先说onNext。它是流式数据的通道，可以被调用任意次，每一次调用都代表向客户端推送一条消息。服务端流里，它就是唯一的"出口"，所有中间结果都从这里出去。注意，onNext不是线程安全的，如果你在服务端用多个线程同时调用同一个StreamObserver实例的onNext，轻则报错，重则消息交错乱序，这个后面我会专门讲。

再说onError和onCompleted。这两个方法都代表流的终结，而且是互斥的，只能二选一调用一次。调用了onError，客户端会收到一个携带着错误状态码和描述的异常；调用了onCompleted，客户端会收到一个"流正常结束"的信号。关键禁忌是：一旦调用了其中一个，就不能再调用onNext，也不能再调用另一个，否则gRPC框架会直接抛异常，连接状态也会被破坏。

下面这个状态流转图我用文字描述一下，大家对照理解：流一开始是活跃状态，可以任意次调用onNext；当业务处理遇到无法恢复的异常时，进入终止态并调用onError；当所有数据都推完时，调用onCompleted进入正常终止态。两个终止态都是终态，没有回头路。

2.2 四种调用模式：StreamObserver各自怎么参与

gRPC一共支持四种调用模式，StreamObserver在每种模式里的戏份完全不同。

如果我定义一个普通的方法，请求和响应都只有一个，叫一元RPC，这种模式其实用不太到StreamObserver，服务端实现里拿到的是返回值，客户端拿到的是阻塞或异步的Future。但从服务端流式开始，StreamObserver就登场了。

我用表格把这四种模式对照一下，方便大家记忆：

这里容易绕晕的是客户端流和双向流：服务端方法的入参和返回值同时出现了两个StreamObserver。以双向流为例，入参的那个Observer是服务端用来"读"客户端消息的，客户端每发一条消息，gRPC就会回调这个Observer的onNext；返回值那个Observer是服务端用来"写"给客户端的通道。一进一出，互不干扰。

实际做AI推理服务，最常用的是服务端流和双向流。大模型对话用服务端流就够，客户端发完请求坐等接收。但如果要做带中断反馈的对话（比如用户点了"停止生成"，客户端立刻发一个取消指令，同时服务端还在继续推token），双向流更顺手，因为客户端可以在同一个流上把控制指令发过去，服务端收到后立即终止本轮生成。

2.3 为什么不用Future/CompletableFuture拿结果

很多第一次接触StreamObserver的人会问：客户端接收流式响应的时候，能不能像CompletableFuture那样，等所有数据到齐了再统一处理？理论上可以，你把收到的消息攒进一个List，等onCompleted之后一起处理就行。但这就把流式传输硬生生变成了批量传输，那你还不如直接用一元RPC。

Future的模型本质是在等"一个最终值"，它表达不了"值会分多次到达"这层语义。你想象一下：点菜时告诉服务员"等整桌菜上齐了再叫我"，跟"每上一道菜就喊我一声"的区别。Future是前者，StreamObserver是后者。AI场景里，模型输出是渐次产生的，服务端生成完第一个token就想立刻让客户端看到，只有推式回调能做到，Future只能干等。

另一点，StreamObserver是异步的。客户端调用Stub方法时不会阻塞当前线程，消息到达后由gRPC的线程池调度onNext回调。这意味着你可以用少量线程管理大量在途请求，这在AI推理这种"请求多、单个耗时长"的场景特别重要。如果每个请求都占一个线程死等，线程池很快就会被占满，吞吐量直接腰斩。

3. 实操：在AI推理服务中落地StreamObserver

3.1 先定好proto契约

我下面用一个典型的LLM流式对话接口来演示。假设我们有一个推理服务，客户端传入prompt，服务端不断把生成出来的token片段推回来。

syntax = "proto3"; package ai.inference.v1; service LLMService { rpc Chat(ChatRequest) returns (stream ChatResponse); } message ChatRequest { string prompt = 1; int32 max_tokens = 2; double temperature = 3; string session_id = 4; } message ChatResponse { string token = 1; int64 sequence = 2; string finish_reason = 3; // 空表示还没结束，normal表示正常结束，stop表示用户停止 }

注意proto里写的是returns (stream ChatResponse)，这个stream关键字一加，生成的代码就完全不同了：服务端接口不再返回ChatResponse，而是接收一个StreamObserver<ChatResponse>参数；客户端Stub对应的方法也要求额外传一个StreamObserver<ChatResponse>。

字段设计上我多说一句。流式消息里一定要带sequence这个字段，因为底层的HTTP/2多路复用和gRPC调度虽然多数时候能保证顺序，但一旦出错排查起来非常困难，带一个序号字段，客户端可以做乱序检测和去重，成本极低，收益极大。还有finish_reason这个字段，业务上必须要有，客户端靠它判断流是"干净地结束"（比如输出了正常终止词）还是"被中途掐断"（比如超时或模型显存溢出）。

3.2 服务端实现：往Observer上推token

proto编译后，我们的服务类需要实现LLMServiceGrpc.LLMServiceImplBase，重点看Chat方法。

@Override public void chat(ChatRequest request, StreamObserver<ChatResponse> responseObserver) { // 注意：这里不能阻塞当前线程太久，gRPC会把它当作worker线程占用 inferenceExecutor.submit(() -> { try { String sessionId = request.getSessionId(); SequenceGenerator seq = new SequenceGenerator(); // 模拟大模型逐token生成 for (String token : model.generate(request.getPrompt(), request.getMaxTokens())) { ChatResponse resp = ChatResponse.newBuilder() .setToken(token) .setSequence(seq.next()) .build(); responseObserver.onNext(resp); // 模拟真实推理的耗时，让客户端能看到渐进效果 Thread.sleep(80); } ChatResponse done = ChatResponse.newBuilder() .setFinishReason("normal") .build(); responseObserver.onNext(done); responseObserver.onCompleted(); } catch (Exception e) { // 任何异常都必须转为onError，否则客户端永远等不到结果 responseObserver.onError(Status.INTERNAL .withDescription("inference failed: " + e.getMessage()) .asRuntimeException()); } }); }

这里有几个非常关键的实操点。第一，不要直接在gRPC的worker线程里跑推理，否则一个慢推理就会拖垮整个Netty的eventLoop，其他请求全部跟着遭殃。我上面专门用inferenceExecutor把实际推理扔到了独立线程池。第二，生成过程中每个token立即通过onNext推送，不要攒批，攒批会让TTFT重新变高。第三，异常处理必须完整，任何中途异常都要转成onError推给客户端，否则客户端那边的Observer会一直挂在那里直到超时，这个是最常见的线上问题。

有一点我要强调：上面代码里的Thread.sleep(80)是我为了演示加的人工延迟。真实项目中，如果模型服务本身是流式返回的，服务端应该把模型框架返回的每一条结果直接透传到responseObserver上，结构大致是：gRPC收到流式响应 -> 逐条封装成ChatResponse -> onNext推给客户端 -> 模型流结束 -> onCompleted。核心思想是"不加工、不攒批、不入队，模型怎么吐，Observer就怎么推"。

3.3 客户端接入：实现自己的StreamObserver

客户端这边，核心就是实现一个StreamObserver，然后把回调写进业务逻辑里。

public class ChatClient { private final LLMServiceGrpc.LLMServiceStub stub; public ChatClient(ManagedChannel channel) { this.stub = LLMServiceGrpc.newStub(channel); } public void chat(String prompt) { StreamObserver<ChatResponse> responseObserver = new StreamObserver<>() { @Override public void onNext(ChatResponse resp) { // 每个token到达时都会被回调，这里直接渲染到前端 if (!resp.getFinishReason().isEmpty()) { System.out.println("[流结束] reason=" + resp.getFinishReason()); } else { System.out.print(resp.getToken()); System.out.flush(); } } @Override public void onError(Throwable t) { System.err.println("调用失败: " + t.getMessage()); } @Override public void onCompleted() { System.out.println("[对话完成]"); } }; ChatRequest request = ChatRequest.newBuilder() .setPrompt(prompt) .setMaxTokens(512) .setTemperature(0.7) .build(); stub.chat(request, responseObserver); // 注意：此处方法立即返回，真正的逻辑全在回调里 } }

这里要特别提醒一点：stub.chat(request, responseObserver)是异步的，调用完就立刻返回了，代码走到这里并不会阻塞等待结果。真正的业务逻辑都在responseObserver的回调里。如果你对这个异步模型不熟悉，很容易写出"方法调用完下一步就能用到结果"的错误代码。

回调里的线程也不是你调用chat方法的那个线程，而是gRPC内部的调度线程。所以如果你的业务逻辑涉及UI刷新、状态管理之类的线程敏感操作，一定要在回调里切换到对应线程后再处理，不能直接在onNext里操作非线程安全的数据结构。

3.4 双向流与背压控制

双向流更贴近AI实时交互的真实形态，客户端可以随时插入控制消息，服务端也可以持续推送推理结果。proto定义如下：

rpc ChatStream(stream ChatStreamRequest) returns (stream ChatStreamResponse); message ChatStreamRequest { oneof payload { string prompt = 1; // 发送新对话 string stop = 2; // 请求停止当前生成 } } message ChatStreamResponse { string token = 1; int64 sequence = 2; string event = 3; // token / start / stop_ack }

服务端实现里，入参的StreamObserver<ChatStreamRequest>负责接收客户端消息，返回值的StreamObserver<ChatStreamResponse>负责推送给客户端。当客户端发来stop消息，服务端可以立刻终止当前模型推理，并在响应流里回一个stop_ack事件，让客户端确定"服务端已经收到停止指令"，避免出现"客户端以为停了、服务端还在偷偷生成"的尴尬。

背压这块，很多人会忽视。gRPC的流式传输底层是有流控的，HTTP/2在帧层面有一个flow control window，默认通常是1MB左右。如果客户端消费慢、服务端生产快，窗口被填满后发送端会被阻塞，这就是天然的背压。这意味着你不需要在应用层堆一个"待发送队列"，因为底层已经帮你按TCP窗口做了限速。

但这里有个实用经验：如果你明确知道某些响应消息很大（比如一次推送几百KB的向量数据），可以考虑调大HTTP/2的流控窗口，减少小窗口频繁确认带来的吞吐损耗。在grpc-java里可以通过NettyChannelBuilder设置：

NettyChannelBuilder.forAddress(host, port) .flowControlWindow(4 * 1024 * 1024) // 4MB .build();

服务端也可以在NettyServerBuilder里设置同样的参数。调大窗口要付出的代价是单个连接上内存占用上升，所以不是越大越好，通常4MB到8MB是比较合理的区间，适合媒体流和向量流场景。

4. 高频报错的排查实录

刚才讲的是正确用法，现在来说说实际运维里遇到的那些报错。这些报错我基本都在生产环境见过，有的折腾了我一整天，拿出来给各位排雷。

4.1 cannot finish rpc call in 30 seconds: null

这个报错基本可以翻译为"RPC调用在30秒内没有完成，且拿不到具体的错误状态"。它的出现通常意味着请求被服务端挂住了，或者网络链路迟迟不给最终响应。排查这个报错，我按下面的顺序来。

先确认是不是客户端自己设置了超时。grpc-java里可以通过withDeadlineAfter设置Deadline，如果没设置，默认情况下很多场景没有硬超时，但如果你在代码里加了30秒的deadline，服务端一旦超过这个时间没返回最终状态，客户端就会主动抛这个错。解决办法很简单：要么把deadline放宽（比如AI推理的场景我建议60到120秒起步），要么把推理改成流式传输，让服务端先返回部分结果"占住"，避免中间链路误判超时。

再看服务端是不是处理不过来。排查时先看服务端的线程池指标有没有被打满。之前我遇到过一次，模型推理线程池核心线程数设置太小，推理请求排着长队，第一批请求等了几十秒都没进模型，客户端那边早就超时了。这时候光调客户端超时是没用的，得增加服务端的并发处理能力，或者引入排队机制并明确告知客户端"你被排到第几位了"。

还有可能是中间链路的问题。如果服务端前面挂了网关、负载均衡器，它们各自的响应超时设置也要一并检查。有一次我排查了很久，最后发现是网关的negotiation timeout设了30秒，而推理刚好要35秒，网关直接把连接掐了。这个排查思路是：从客户端到服务端，每一跳的超时设置都列出来，找到最短的那一个，它基本就是罪魁祸首。

4.2 curl 56 schannel: server closed abruptly / openssl ssl_read error

这类报错的完整形态通常是这样的：error: rpc failed; curl 56 schannel: server closed abruptly (missing close_notify)，在Linux环境下则表现为curl 56 openssl ssl_read: error:1408f119:ssl routines:ssl3_get_record:decryption failed or bad record mac。注意，这不是gRPC场景的报错，它通常出现在使用git或某些RPC协议客户端通过HTTP拉取数据时。

两个报错的核心原因很相似：TCP连接在正常关闭之前，TLS层没有收到预期的close_notify告警，服务端就强行断了连接。Schannel是Windows上的TLS实现，OpenSSL是Linux和macOS上的，所以同一类问题在不同平台上报错内容看起来完全不同，本质都是"连接被服务端异常关闭"。

排查这类问题，我建议从这几个方向入手。第一，看看是不是有防火墙或中间代理在静默掐连接。很多企业网络出口的防火墙会设置空闲连接超时，连接空闲超过一定时间就被标记为僵尸连接清掉，客户端再往这个连接上发数据，自然就撞上了"server closed abruptly"。第二，检查服务端的keepalive配置。如果你在nginx或Spring服务端配了keepalive_timeout 0或很小的值，连接很快就被服务端主动关了，客户端还没反应过来。第三，如果客户端是你自己写的，可以检查一下HTTP版本协商。有些老旧的TLS栈和新的HTTP/2服务端配合不好，尝试强制使用HTTP/1.1或关闭连接复用可能会绕过这个报错。

4.3 error grabbing logs: rpc error code = unknown desc = warning: incomplete log

这个报错我是在Kubernetes环境里遇到的，形式一般是error grabbing logs: rpc error: code = Unknown desc = warning: incomplete log。它出现在kubectl logs拉取容器日志时，底层是节点上的容器运行时通过CRI接口的另一套RPC（Container Runtime Interface，简称CRI）把日志流式返回给kubelet，中间某个环节日志传输不完整，于是CRI层的RPC返回了一个Unknown错误。

出现这种情况，八成是容器日志文件正在被高频写入，而日志轮转（log rotation）恰好把当前正在读取的文件给截断了。解决思路有三条：第一，调整容器运行时的日志轮转策略，让日志文件更大一些，轮转频率低一些；第二，把kubelet的日志拉取请求换成一个"稳定副本"，也就是从日志的持久化存储里取，而不是直接从节点上抓实时文件；第三，如果是偶发报错，重试一次大概率就成功了，因为下一次读取可能恰好绕过了截断窗口。

这里顺便说一句，"RPC"这个词在不同技术栈里指代不同东西，排查时要先定位你遇到的是哪个层面的RPC，是gRPC、基础设施的CRI，还是别的，千万别混着看。

4.4 被名字坑了的另一个RPC：GDAL里的RPC正射校正

提到RPC，不得不提醒一句，在地理信息系统领域，RPC是另一个完全不同的缩写：Rational Polynomial Coefficients，有理多项式系数。做遥感影像处理的人经常看到"RPC正射校正"这个说法，它和"远程过程调用"没有半毛钱关系。

GDAL做正射校正时用RPC文件里的系数来建立影像像素坐标和地面坐标之间的映射关系，再配合UTM投影做几何校正。这里就出现了搜索引擎里常见的组合关键词："gdal rpc正射校正utm投影安装步骤与注意事项"。如果你是因为搜RPC相关的技术问题被带到了这篇文章，先确认一下你要做的是不是影像校正。如果是，那重点就不在StreamObserver，而在GDAL的版本兼容性和RPC文件的正确性上。

GDAL做RPC正射校正的安装过程里，最容易翻车的是PROJ库版本不匹配。建议在conda环境里用conda install -c conda-forge gdal proj一次装完，避免系统自带的老版本PROJ导致坐标转换出现莫名其妙的偏移。校正之前一定要先检查RPC文件里的坐标系元数据，对照影像的投影信息，尤其是UTM分带，很多校出来的影像位置偏到海里，就是因为UTM带号没对上。

5. 工程化避坑心得

5.1 onNext回调里别做重活

这点我必须放在最前面说。StreamObserver的回调发生在gRPC的Netty线程上，如果你在onNext里执行耗时操作，比如查数据库、调外部接口、做大量JSON解析，这些线程就会被占住，直接影响整个gRPC连接上的消息收发。多个请求会因此互相拖累。

正确做法是在onNext里只做轻量处理和转发，把真正耗时的事情丢到业务线程池。举个例子，我在一个实时音频转写服务里，客户端会高频收到音频片段的转写文本，服务端把文本透传到客户端之后，客户端需要把文本写入消息队列和实时字幕渲染。这两个操作都不适合直接写在onNext里，所以我当时是把它投递到了一个基于Disruptor的无损队列里，再由独立的render线程消费。

5.2 忘记onError的后果比你想的严重

服务端代码里，如果某个分支发生了异常但没有调用onError，也没有调用onCompleted，那意味着这个流的终止信号永远不会到达客户端。客户端会一直在那儿等，直到超时。如果是长连接复用，这个半死不活的流还会继续占用连接资源，越积越多，最后把整个Netty连接池拖垮。

我养成了一个习惯：服务端所有流式方法的入口和出口用try-with-resources或者CompletableFuture的whenComplete来兜底，保证任何异常路径都必然走onError，任何正常路径都必然走onCompleted。写了一个小封装，类似这样：

private void finishStream(StreamObserver<?> observer, Throwable error) { if (error != null) { observer.onError(Status.UNKNOWN.withDescription(error.getMessage()).asRuntimeException()); } else { observer.onCompleted(); } }

所有流式方法末尾统一调用，就能杜绝"流悬挂"这个问题。

5.3 客户端取消后，服务端要能感知

这是一个容易被忽视的配合问题。客户端因为超时、用户手动停止等原因，主动取消了请求，但服务端如果还在傻乎乎地推理、往onNext上推数据，那这条流上的数据就全部被丢弃了，白白浪费算力。更麻烦的是，如果客户端取消时服务端的流没有被终止，后续再次调用onNext会抛异常，服务端需要正确处理这个异常，否则日志里会大量刷错。

grpc-java提供了Context机制，服务端可以通过监听Context.current().isCancelled()来感知客户端取消。常用的做法是在进入流式方法时注册一个回调：

Context.current().addListener(context -> { // 客户端取消了，立刻中断模型推理线程 inferenceTask.cancel(true); }, directExecutor());

这样客户端一取消，服务端就能立刻把推理停掉，释放GPU和内存资源。尤其在长文本生成场景，这个动作能省下大量无效计算。

5.4 流控窗口与内存水位要配套调

前面提到可以调大HTTP/2流控窗口来提高吞吐，但窗口调大之后，单个连接的缓冲区内存占用也随之上升。尤其是你服务端同时挂了几百个流式请求时，每个流的发送窗口都很大，Netty的堆外内存可能会先被撑爆。

我的调优经验是：先按"单流消息大小 x 预期并发流数 x 2"估算出大致的内存占用，再决定窗口大小。例如每条消息平均64KB，预期同时活跃500个流，那窗口大小设为1MB会在极端情况下占满约500MB的缓冲，如果部署的Pod内存只有2GB，这个值就要再降。内存和吞吐永远是一对矛盾，越是大流量越要精细核算，不能拍脑袋调参数。

5.5 流式消息里埋traceId和sequence

前面写proto时我强调了sequence字段。这里再补充一个：流式接口强烈建议在消息里带traceId、sequence两个字段。traceId用于串联调用链，排查问题时可以通过日志链路查到某个token是哪次请求产生的；sequence用于检验消息顺序和完整性。有一次我遇到消息乱序，是客户端接收线程池配置了多线程导致onNext回调和业务处理之间产生了竞态。如果没有sequence，这个问题我可能查一个通宵，有了sequence，客户端一比对序号就立刻发现了乱序点。

这个做法的成本几乎为零，但在排障时的价值无可估量。做流式接口的同学，把traceId和sequence加进proto，是我能给出的最实在的建议。

结尾

我做gRPC流式传输这几年，最大的体会是：流式方案的难点其实不在"怎么把数据推出去"，而在"怎么让对端在正确的时机收到正确的数据，并且双向都能优雅地结束"。StreamObserver把底层HTTP/2的流式语义包装成了三个回调方法，用它做AI推理服务的传输层非常顺手，但前提是你得真正理解它的生命周期约束：onNext可以很多次，onError和onCompleted只能一次，终止之后一切都结束。

如果你正在设计一个新的AI服务接口，我的建议是：只要响应可能超过几MB，或者生成时间可能超过几秒，直接上服务端流式，别犹豫。把流控窗口、超时时间、取消传播这几件事在架构阶段就想清楚，不要等到线上出了OOM或者流悬挂了再回头补。最后再分享一个小技巧：在流式接口的每一条消息里都带上序号和traceId，哪怕一开始觉得用不上，等你需要排查线上问题时，会发现它们是救命的。