java - 使用多个 ResponseObserver.onNext 会卡住服务 - gRPC Java

Question

我想使用 gRPC 以 50 个 block 的形式发送数据，但是多次运行 responseObserver.onNext 时，它会卡住并且不会发送数据。但是当我停止流时，我得到了数据。

这是我的代码:

// List<List<MyClass>> listOfMyClass
for (List<MyClass> myClasses : listOfMyClass) {
  responseObserver.onNext(buildReply(myClasses));
}
responseObserver.onCompleted();

这会使其卡住。但是，如果我只运行 responseObserver.onNext(buildReply(myClasses)); 一次，我将立即获取数据。

我的原型(prototype):

message Request {
    string number = 1;
}

message Reply {
    repeated CustomMessage results = 1;
}

service Service {
    rpc MyRequest (Request) returns (stream Reply) {}
}

我一直在使用名为 https://github.com/uw-labs/bloomrpc 的 GUI ，这应该可以轻松显示流。

Answer 1

StreamObserver 接口(interface)是非阻塞的，不会为出站数据提供背压。为了给客户端提供反压，需要将客户端用来发送数据的StreamObserver强制转换为ServerCallStreamObserver，并观察“ready”位。像这样的事情:

ClientCallStreamObserver<MyClass> clientCallStreamResponseObserver =
      (ClientCallStreamObserver<File>) responseObserver;
clientCallStreamObserver.setOnReadyHandler(new Runnable() {
    public void run() {
      sendUntilNotReady(clientCallStreamResponseObserver);
    }
});
sendUntilNotReady(clientCallStreamResponseObserver);

...

private void sendUntilNotReady(ClientCallStreamObserver<MyClass> clientCallStreamResponseObserver) {
  while (clientCallStreamResponseObserver.isReady()) {
    clientCallStreamResponseObserver.onNext(... /* build reply for next item in listOfMyClass */);
  }
}