承接的问题
前两章解决「消息长什么样」和「怎么改」。到了调用侧，第一个决策是选哪种调用模式。很多团队默认所有方法都用 unary，遇到大结果集就分页，遇到上传就切片，最后用一堆业务字段模拟流。这样做能跑，但把流控、取消和内存管理都推给了业务代码。
gRPC 提供四种模式：unary、server streaming、client streaming、bidi streaming。它们不是语法糖，而是不同的流生命周期：谁先半关闭、响应什么时候开始、取消怎么传播、流控由谁负责。选错模式，后面要用更多代码补回来。
四种模式的分工
| 模式 | MethodType | 客户端发 | 服务端回 | 典型场景 |
|---|---|---|---|---|
| Unary | UNARY | 1 条 | 1 条 | 查询、创建、更新、删除 |
| Server streaming | SERVER_STREAMING | 1 条 | N 条 | 大结果集、订阅、进度推送 |
| Client streaming | CLIENT_STREAMING | N 条 | 1 条 | 上传、批量聚合、累计计算 |
| Bidi streaming | BIDI_STREAMING | N 条 | N 条 | 聊天、实时协作、双向协商 |
linkd-v3 的 Library 服务四个方法刚好覆盖四种：
service Library {
rpc GetBook(GetBookRequest) returns (Book); // unary
rpc SearchBooks(SearchBooksRequest) returns (stream Book); // server streaming
rpc WatchStock(WatchStockRequest) returns (stream StockUpdate); // server streaming
rpc Chat(stream ChatMessage) returns (stream ChatMessage); // bidi
}
这里没有 client streaming 的方法。不是它没用，而是当前业务没有「客户端连续上传、服务端最后回一个汇总」的场景。如果要做批量导入，ImportBooks(stream Book) returns (ImportResult) 就是典型的 client streaming。
选择模式时看三个问题：
- 响应是不是一条？ 是 → unary 或 client streaming；否 → server streaming 或 bidi。
- 请求是不是一条？ 是 → unary 或 server streaming；否 → client streaming 或 bidi。
- 请求和响应需不需要交替？ 需要 → bidi；不需要 → 单向流。
Unary 与 server streaming
Unary 是最熟悉的模式：客户端发一条，服务端回一条，然后流结束。它的语义简单，重试、超时、幂等都好处理。绝大多数 CRUD 都应该用 unary。
Server streaming 是「一个请求，多条响应」。服务端在 onNext 里发多条消息，最后 onCompleted。客户端可以用阻塞式迭代器逐条读，也可以用异步 StreamObserver：
package com.plume.linkd.v3.client;
import com.plume.linkd.v3.api.Book;
import com.plume.linkd.v3.api.LibraryGrpc;
import com.plume.linkd.v3.api.SearchBooksRequest;
import io.grpc.ManagedChannel;
import java.util.Iterator;
public final class SearchClient {
public static void main(String[] args) {
ManagedChannel channel = io.grpc.ManagedChannelBuilder
.forAddress("127.0.0.1", 9093)
.usePlaintext()
.build();
try {
Iterator<Book> books = LibraryGrpc.newBlockingStub(channel)
.withDeadlineAfter(800, java.util.concurrent.TimeUnit.MILLISECONDS)
.searchBooks(SearchBooksRequest.newBuilder()
.setQuery("java")
.setPageSize(20)
.build());
while (books.hasNext()) {
Book book = books.next();
System.out.println("book=" + book.getIsbn() + " " + book.getTitle());
}
} finally {
channel.shutdownNow();
}
}
}
阻塞式迭代器的 hasNext() 会等到下一条消息或流结束。它适合批处理、离线任务和命令行工具，不适合在事件循环线程里用：等待期间线程被占住，无法处理其它调用。服务端实现里不要调用阻塞 API，客户端在 UI 线程或事件循环里也不要用。
Server streaming 的三个边界：
- 服务端发完必须
onCompleted。 只发onNext不结束，客户端会一直等，直到 deadline 触发DEADLINE_EXCEEDED。 - 客户端可以提前取消。 迭代器
break或关闭 Channel 会发送RST_STREAM，服务端的ServerCallStreamObserver.isCancelled()变为 true。 - 流控按消息计。 客户端消费慢时，服务端的
isReady()会变 false，继续写会堆在缓冲区里。大结果集必须检查isReady()，不能一次onNext一万条。
WatchStock 也是 server streaming，但语义不同：它不是「一次查询的完整结果」，而是「持续推送直到取消」。这种流通常没有自然结束，靠 deadline 或客户端取消收敛。服务端要在 setOnCancelHandler 里清理订阅，否则客户端断开后定时任务还在跑。
Client streaming 与 bidi
Client streaming 是「多条请求，一条响应」。客户端连续 onNext，最后 onCompleted 半关闭；服务端收完后回一条。它适合上传、批量导入和累计聚合。
Bidi streaming 是两个方向都开放。客户端和服务端各自维护一个 StreamObserver，可以交替发送。注意：gRPC 不保证「发一条就回一条」的配对语义，只保证每个方向内部有序。需要请求-响应配对时，要在消息里带关联 ID。
Library.Chat 的回显实现和客户端调用如下。服务端把收到的每条消息原样回发，客户端发两条后关闭：
package com.plume.linkd.v3.server;
import com.plume.linkd.v3.api.Book;
import com.plume.linkd.v3.api.ChatMessage;
import com.plume.linkd.v3.api.GetBookRequest;
import com.plume.linkd.v3.api.LibraryGrpc;
import com.plume.linkd.v3.api.SearchBooksRequest;
import com.plume.linkd.v3.api.StockUpdate;
import com.plume.linkd.v3.api.WatchStockRequest;
import io.grpc.stub.StreamObserver;
public final class LibraryService extends LibraryGrpc.LibraryImplBase {
@Override
public void getBook(GetBookRequest request, StreamObserver<Book> responseObserver) {
responseObserver.onNext(Book.newBuilder()
.setIsbn(request.getIsbn())
.setTitle("book:" + request.getIsbn())
.setStock(1)
.build());
responseObserver.onCompleted();
}
@Override
public void searchBooks(SearchBooksRequest request, StreamObserver<Book> responseObserver) {
for (int i = 0; i < Math.min(request.getPageSize(), 3); i++) {
responseObserver.onNext(Book.newBuilder()
.setIsbn("q-" + i)
.setTitle(request.getQuery() + "-" + i)
.build());
}
responseObserver.onCompleted();
}
@Override
public void watchStock(WatchStockRequest request, StreamObserver<StockUpdate> responseObserver) {
responseObserver.onNext(StockUpdate.newBuilder()
.setIsbn(request.getIsbn())
.setStock(1)
.setVersion(1)
.build());
responseObserver.onCompleted();
}
@Override
public StreamObserver<ChatMessage> chat(StreamObserver<ChatMessage> responseObserver) {
return new StreamObserver<>() {
@Override
public void onNext(ChatMessage message) {
responseObserver.onNext(message);
}
@Override
public void onError(Throwable throwable) {
responseObserver.onError(throwable);
}
@Override
public void onCompleted() {
responseObserver.onCompleted();
}
};
}
}
watchStock 示例只推一条就结束，是为了让片段能独立运行；真实的库存订阅会在取消前持续推送，靠 setOnCancelHandler 清理定时任务和订阅。
客户端用异步 stub，因为 bidi 没有阻塞式调用：
package com.plume.linkd.v3.client;
import com.plume.linkd.v3.api.ChatMessage;
import com.plume.linkd.v3.api.LibraryGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public final class ChatClient {
public static void main(String[] args) throws Exception {
ManagedChannel channel = ManagedChannelBuilder
.forAddress("127.0.0.1", 9093)
.usePlaintext()
.build();
CountDownLatch done = new CountDownLatch(1);
try {
StreamObserver<ChatMessage> requests = LibraryGrpc.newStub(channel).chat(
new StreamObserver<>() {
@Override
public void onNext(ChatMessage message) {
System.out.println("stream=" + message.getText());
}
@Override
public void onError(Throwable throwable) {
System.out.println("error=" + throwable.getMessage());
done.countDown();
}
@Override
public void onCompleted() {
System.out.println("stream-complete");
done.countDown();
}
});
requests.onNext(ChatMessage.newBuilder().setFrom("client").setText("echo:a").build());
requests.onNext(ChatMessage.newBuilder().setFrom("client").setText("echo:b").build());
requests.onCompleted();
done.await(1, TimeUnit.SECONDS);
} finally {
channel.shutdownNow();
}
}
}
本机用等价的 bidi 流式服务定义验证过同样的收发顺序（Temurin 25.0.5+8，grpc-netty-shaded 1.76.0），输出：
stream=echo:a
stream=echo:b
stream-complete
这个输出确认了三件事：两个方向的流都能传消息、服务端按收到的顺序回显、onCompleted 从客户端传到服务端再传回客户端。
Bidi 的常见误解是「服务端可以随时主动发消息」。可以，但前提是客户端还持有这个流并且没有半关闭。客户端 onCompleted 之后，服务端再 onNext 会失败；需要长期推送时，用 server streaming 或让客户端保持流打开。
在 linkd-v3 里落地：四个方法一起实现
LibraryService 的骨架已经在上面给出。落地时还有几件事要补：
| 事项 | 做法 | 原因 |
|---|---|---|
| 服务注册 | ServerBuilder.addService(new LibraryService()) | 生成代码实现 BindableService |
| 拦截器 | 请求 ID、租户、deadline | 3.2 和 4.2 展开 |
| 健康检查 | HealthStatusManager | 5.3 和 15.2 使用 |
| 反射 | ProtoReflectionService | 调试用，grpcurl 可选 |
| 流控 | ServerCallStreamObserver.isReady() | 大结果集必须检查 |
| 取消清理 | setOnCancelHandler | 订阅、定时任务、数据库游标 |
注册代码本身很短：
Server server = ServerBuilder.forPort(9093)
.addService(new LibraryService())
.addService(new HealthStatusManager().getHealthService())
.addService(ProtoReflectionService.newInstance())
.maxInboundMessageSize(4 * 1024 * 1024)
.build()
.start();
maxInboundMessageSize 和服务端默认值可能不同，显式写 4 MiB 是为了和全书口径一致。ProtoReflectionService 让支持反射的客户端能列出方法；本机没有 grpcurl，它的验证放在 15.2 用 Java 代码完成。
流控、线程与取消
流式调用最容易出问题的地方不是语法，而是线程和流控：
- 不要在服务端方法里阻塞。
getBook的签名是同步的，但实现里如果调用阻塞 IO，会占住 gRPC 的调用线程。用独立的业务线程池，或者用异步 API。 - 检查
isReady()。ServerCallStreamObserver.isReady()为 false 时，继续onNext会把消息堆在内存里。正确做法是注册setOnReadyHandler，在可写时继续发。 - 手动流控要配对。
disableAutoRequest()之后必须自己request(n)，否则一条消息都收不到。这个 API 适合需要按批处理的场景，不适合随手打开。 - 取消要清理。 客户端断开后，服务端的
isCancelled()变 true。数据库游标、订阅、定时器都要在setOnCancelHandler里释放。 - 半关闭不是关闭。 客户端
onCompleted只表示「我不再发了」，仍然可以接收响应。服务端收到onCompleted后应该完成响应并onCompleted，不要一直等新消息。
流控的单位是消息，不是字节。一条 4 MiB 的消息和一条 4 字节的消息都算一条；按消息数限流不能代替按字节限流，反之亦然。8.3 会把这两者放在一起讨论。
常见错误与边界
错误一：所有接口都用 unary。 大结果集靠分页模拟流，客户端要发多次请求，服务端要重复查询；分页之间数据变化还会导致重复或遗漏。真正需要「一次请求、持续结果」时用 server streaming。
错误二：server streaming 不发送 onCompleted。 客户端会一直等到 deadline。即使业务上「流不会结束」，也要在取消或错误时明确结束。
错误三：bidi 里假设请求响应一一对应。 gRPC 只保证单向有序，不保证配对。需要配对时在消息里带 request_id，或者改用 unary。
错误四：客户端在 UI 线程用阻塞迭代器。 hasNext() 会阻塞，界面会卡住。UI 场景用异步 stub，把回调切回 UI 线程。
错误五：服务端在 onNext 里做重活。 onNext 在调用线程上执行，耗时操作会拖慢整个流。把处理放到业务线程池，注意保持消息顺序。
边界：client streaming 的错误返回只有一条。 服务端在收完所有消息后回一条响应；如果中途失败，只能 onError，已经发送的响应无法撤回。
边界：流式调用也受 4 MiB 单条消息限制。 每条消息独立计算，不是整个流的总大小。大文件要切块，块大小留出编码和压缩余量。
边界：取消不一定立刻到达服务端。 RST_STREAM 可能因为网络延迟晚到，服务端的 isCancelled() 有滞后。不要依赖「取消后立刻停止计费」这类强语义。
小结
- 四种模式的分工：unary 一问一答，server streaming 一问多答，client streaming 多问一答，bidi 多问多答。
- 选择看两个问题：响应是否多条、请求是否多条；需要交替时用 bidi。
Library的四个方法覆盖 unary、两种 server streaming 和 bidi；client streaming 留给批量导入场景。- 服务端发完必须
onCompleted；客户端可以提前取消，服务端要在setOnCancelHandler里清理。 - 流控按消息计，
isReady()为 false 时不要继续写；disableAutoRequest()后必须手动request(n)。 - bidi 不保证请求响应配对，只保证单向有序；需要配对时在消息里带关联 ID。
- 本机 bidi 输出为
stream=echo:a、stream=echo:b、stream-complete。
调用模式选定后，最容易被忽略的是「这次调用最多能花多久」。3.2 从 800 ms 总预算和 300 ms 单跳预算讲起，解释 deadline 怎么传播、取消怎么收敛、剩余时间怎么在重试之间分配。
阅读导航：上一节 2.3 字段演进与兼容矩阵 ｜ 下一节 3.2 Deadline 传播与取消
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。