JAVA GRPC 服务器端流不断增加 max-inbound-message-size

Question

使用 GRPC 传输大量数据的最佳实践是什么？我正在向 GRPC 服务器发送请求，该服务器将流回数据。发回的数据可以是大约 100 个 protobuf 消息，也可以是几个 100.000 个 protobuf 消息。

service CrossEngineSelector {
    rpc QueryDB (QueryRequest) returns (stream QueryResponse) {}
}

服务器是一个发送 protobuf 消息的简单实现。

@Override
public void queryBD(QueryRequest request, StreamObserver<QueryResponse> responseObserver) {

      Iterables.partition( dataLoader.getData(), 1000).forEach(batch -> {
          responseObserver.onNext(QueryResponse.newBuilder().addAllRows(batch).build());
      });
      responseObserver.onCompleted();
}

客户端使用一个blockingStub来调用这个方法(由protobuf生成的代码):

public Iterator<QueryResponse> queryDB(QueryRequest request) {
            return ClientCalls.blockingServerStreamingCall(this.getChannel(),
                   CrossEngineSelectorGrpc.getQueryEnginesMethod(),
                   this.getCallOptions(), request);
}

一旦客户端调用此方法，我只需迭代 QueryResponse。

所有这些对于仅发回少量消息的流来说都可以正常工作。当我尝试传输 100.000 条消息时，最大入站消息大小不断增加，最终出现错误:RESOURCE_EXHAUSTED:压缩的 gRPC 消息超出最大大小 4194304:读取 4196022 字节

我当前对此的解决方法是将 max-inbound-message-size 设置为非常高的 +1Gb。这是一个硬编码值，因此无法缩放。客户端不知道服务器将返回多少消息。我可能会遇到即使 1Gb 最大入站消息大小也不够的用例。

我希望我犯了一个实现错误。我希望有一种方法可以重置来自服务器的每个流(onNext())的消息大小，或者它不断增加消息大小是否正常？

我假设单个 responseObserver.onNext(QueryResponse.newBuilder().addAllRows(batch).build()); 发送几个 Mb，并且这将被视为消息大小，而不是整个流，只要它运行。

我在服务器和客户端上都使用 Micronaut。

Answer 1

maxInboundMessageSize 用于保护接收方在恶意对等方发送非常大的负载进行攻击时内存不足。这里真正的问题是 1000 个 block 可能大于 4MB(默认 maxInboundMessageSize)。

您可以通过两种方式解决此问题，

• 修复发送方以限制有效负载的大小，而不是行数。
• 或者，接收方有足够大的 maxInboundMessageSize 来处理正常的 1000 行数据。

后面的方法本质上是您实现的。但它应该低于 1GB+。正如我上面提到的，gRPC 不会使用 maxInboundMessageSize 内存量，这纯粹是为了保护。 (稍后添加)