GRPC：用 Java/Scala 制作高吞吐量客户端答案

【问题标题】：GRPC: make high-throughput client in Java/ScalaGRPC：用 Java/Scala 制作高吞吐量客户端
【发布时间】：2020-03-04 23:42:32
【问题描述】：

我有一项服务以相当高的速率传输消息。

目前它由 akka-tcp 提供服务，每分钟发送 350 万条消息。我决定试一试grpc。不幸的是，它导致吞吐量小得多：每分钟约 50 万条消息，甚至更少。

能否请您推荐如何优化它？

我的设置

硬件：32 核，24Gb 堆。

grpc 版本：1.25.0

消息格式和端点

消息基本上是一个二进制 blob。客户端将 100K - 1M 和更多消息流式传输到同一个请求中（异步），服务器不响应任何内容，客户端使用无操作观察器

service MyService {
    rpc send (stream MyMessage) returns (stream DummyResponse);
}

message MyMessage {
    int64 someField = 1;
    bytes payload = 2;  //not huge
}

message DummyResponse {
}

问题：与 akka 实现相比，消息率较低。我观察到 CPU 使用率很低，所以我怀疑 grpc 调用实际上在内部阻塞，尽管它另有说明。调用onNext() 确实不会立即返回，但桌子上还有GC。

我试图产生更多的发件人来缓解这个问题，但没有得到很大的改进。

我的发现 Grpc 在序列化时实际上为每条消息分配了一个 8KB 字节的缓冲区。查看堆栈跟踪：

java.lang.Thread.State: BLOCKED（在对象监视器上）在 com.google.common.io.ByteStreams.createBuffer(ByteStreams.java:58) 在 com.google.common.io.ByteStreams.copy(ByteStreams.java:105) 在 io.grpc.internal.MessageFramer.writeToOutputStream(MessageFramer.java:274) 在 io.grpc.internal.MessageFramer.writeKnownLengthUncompressed(MessageFramer.java:230) 在 io.grpc.internal.MessageFramer.writeUncompressed(MessageFramer.java:168) 在 io.grpc.internal.MessageFramer.writePayload（MessageFramer.java:141）在 io.grpc.internal.AbstractStream.writeMessage(AbstractStream.java:53) 在 io.grpc.internal.ForwardingClientStream.writeMessage（ForwardingClientStream.java:37）在 io.grpc.internal.DelayedStream.writeMessage(DelayedStream.java:252) 在 io.grpc.internal.ClientCallImpl.sendMessageInternal(ClientCallImpl.java:473) 在 io.grpc.internal.ClientCallImpl.sendMessage（ClientCallImpl.java:457）在 io.grpc.ForwardingClientCall.sendMessage（ForwardingClientCall.java:37）在 io.grpc.ForwardingClientCall.sendMessage（ForwardingClientCall.java:37）在 io.grpc.stub.ClientCalls$CallToStreamObserverAdapter.onNext(ClientCalls.java:346)

任何有关构建高吞吐量 grpc 客户端的最佳实践的帮助表示赞赏。

【问题讨论】：

你在使用 Protobuf 吗？仅当 MethodDescriptor.Marshaller.stream() 返回的 InputStream 未实现 Drainable 时，才应采用此代码路径。 Protobuf Marshaller 确实支持 Drainable。如果您使用 Protobuf，是否有可能 ClientInterceptor 正在更改 MethodDescriptor？
@EricAnderson 感谢您的回复。我尝试了带有 gradle 的标准 protobuf (com.google.protobuf:protoc:3.10.1, io.grpc:protoc-gen-grpc-java:1.25.0) 和scalapb。可能这个堆栈跟踪确实是从到 scalapb 生成的代码。我删除了与 scalapb 相关的所有内容，但对性能没有太大帮助。
@EricAnderson 我解决了我的问题。作为 grpc 的开发者，Ping 你。我的回答有意义吗？

标签： java scala grpc

【解决方案1】：

Akka TCP 在这里的表现给我留下了深刻的印象：D

我们的经历略有不同。我们正在使用 Akka Cluster 处理更小的实例。对于 Akka 远程处理，我们使用 Artery 从 Akka TCP 更改为 UDP，实现了更高的速率 + 更低且更稳定的响应时间。 Artery 中甚至还有一个配置可以帮助平衡 CPU 消耗和冷启动的响应时间。

我的建议是使用一些基于 UDP 的框架，该框架还可以为您提供传输可靠性（例如 Artery UDP），并且只使用 Protobuf 进行序列化，而不是使用完整的 gRPC。 HTTP/2 传输通道并不是真正用于高吞吐量低响应时间的目的。

【讨论】：

【解决方案2】：

我通过为每个目标创建多个 ManagedChannel 实例解决了这个问题。尽管有文章说ManagedChannel 本身可以产生足够多的连接，所以一个实例就足够了，但在我的情况下并非如此。

性能与 akka-tcp 实现相当。

【讨论】：

ManagedChannel（具有内置 LB 策略）每个后端不使用多个连接。因此，如果您是高吞吐量且后端很少的情况，则可能会使与所有后端的连接饱和。在这些情况下，使用多个渠道可以提高性能。
@EricAnderson 谢谢。在我的情况下，甚至向单个后端节点生成多个通道也有帮助
后端越少，带宽越高，您就越有可能需要多个通道。因此，“单一后端”更有可能让更多渠道有所帮助。

【解决方案3】：

有趣的问题。计算机网络包使用stack of protocols 编码，此类协议建立在前一个规范之上。因此，协议的性能（吞吐量）受限于用于构建它的协议的性能，因为您在底层协议之上添加了额外的编码/解码步骤。

例如gRPC是建立在HTTP 1.1/2之上的，HTTP 1.1/2是应用层上的一个协议，或者L7，因此它的性能受到@的性能的约束987654326@。现在HTTP 本身是建立在TCP 之上的，它位于传输层 或L4，所以我们可以推断gRPC 吞吐量不能是大于TCP 层中提供的等效代码。

换句话说：如果您的服务器能够处理原始的TCP 包，那么添加新的复杂层 (gRPC) 将如何提高性能？

【讨论】：

正是出于这个原因，我使用了流式传输方法：我为建立一个 http 连接支付了一次费用，并使用它发送了大约 3 亿条消息。它在后台使用 websockets，我希望它的开销相对较低。
对于gRPC，你还需要为建立连接支付一次费用，但是你增加了解析protobuf的额外负担。无论如何，如果没有太多信息，很难做出猜测，但我敢打赌，总的来说，由于您在管道中添加了额外的编码/解码步骤，gRPC 的实现会比等效的网络套接字慢。
Akka 也增加了一些开销。无论如何，x5 减速看起来太多了。
我想你可能会觉得这很有趣：github.com/REASY/akka-http-vs-akka-grpc，在他的情况下（我认为这延伸到你的情况），瓶颈可能是由于 protobuf（反）序列化中的高内存使用，这进而触发对垃圾收集器的更多调用。
只是出于好奇，你到底有什么问题？