本文介绍使用 Guava 的 SettableFuture 将 gRPC 双向流式调用（如 serverReflectionInfo）转为同步阻塞调用，确保在获取响应列表前完整接收所有 onNext 事件，并正确处理完成与异常场景。
本文介绍使用 guava 的 `settablefuture` 将 grpc 双向流式调用（如 `serverreflectioninfo`）转为同步阻塞调用，确保在获取响应列表前完整接收所有 `onnext` 事件，并正确处理完成与异常场景。
在 gRPC 的流式 RPC（如 serverReflectionInfo）中，StreamObserver 的回调方法（onNext、onCompleted、onError）均在 I/O 线程异步执行。若直接在 requestStreamObserver.onCompleted() 后立即访问响应集合，极大概率读到空结果——因为响应尚未抵达或回调尚未调度执行。gRPC 官方不提供内置的同步封装，但可通过轻量级 Future 模式安全桥接异步与同步语义。
推荐方案是使用 Guava 的 SettableFuture（需引入依赖：com.google.guava:guava），它支持手动触发完成状态，天然适配 StreamObserver 的生命周期：
import com.google.common.util.concurrent.SettableFuture;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
// 创建可手动完成的 Future
SettableFuture<list>> responseFuture = SettableFuture.create();
StreamObserver<serverreflectionresponse> responseObserver = new StreamObserver<serverreflectionresponse>() {
private final List<serverreflectionresponse> responses = new ArrayList();
@Override
public void onNext(ServerReflectionResponse response) {
responses.add(response);
}
@Override
public void onError(Throwable t) {
// 关键：传播异常，避免静默失败
responseFuture.setException(t);
}
@Override
public void onCompleted() {
// 所有响应接收完毕，设置最终结果
responseFuture.set(responses);
}
};
// 发起流式调用
StreamObserver<serverreflectionrequest> requestObserver
= reflectionStub.serverReflectionInfo(responseObserver);
requestObserver.onNext(req); // 发送请求
requestObserver.onCompleted(); // 关闭请求流
// 阻塞等待响应完成（可选超时更健壮）
try {
List<serverreflectionresponse> responseList = responseFuture.get(); // 同步获取结果
int sz = responseList.size(); // 此时 size 必然反映真实响应数
return responseList; // 直接作为函数返回值
} catch (ExecutionException e) {
throw new RuntimeException("gRPC call failed", e.getCause());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted while waiting for gRPC response", e);
}</serverreflectionresponse></serverreflectionrequest></serverreflectionresponse></serverreflectionresponse></serverreflectionresponse></list>
⚠️ 注意事项：
-
responseFuture.get()是阻塞调用，请勿在 gRPC I/O 线程或 Android 主线程中使用，否则导致线程挂起；建议在独立业务线程或CompletableFuture.supplyAsync()中调用。 - 生产环境强烈建议使用带超时的
get(long timeout, TimeUnit unit)，防止服务端无响应导致永久阻塞。 - 若需支持多次调用，每次必须新建
SettableFuture和StreamObserver实例，避免状态复用引发竞态。 - 替代方案（如
CountDownLatch或Semaphore）虽可行，但需手动管理完成条件与异常路径，易出错；SettableFuture已封装线程安全、异常传播、重复设置防护等细节，更简洁可靠。
通过该模式，你既能保持 gRPC 的异步高效性，又能在需要同步语义的场景（如单元测试、脚本化工具、遗留同步接口适配）中获得确定性行为。
