如何确保在 gRPC 双向流中收到消息？答案

【问题标题】：How do I make sure that a message was received in gRPC bidirectional streaming?如何确保在 gRPC 双向流中收到消息？
【发布时间】：2022-03-26 05:04:46
【问题描述】：

我如何知道我通过 gRPC 流发送的消息已在另一端收到？

在 gRPC 双向流式传输中是否有内置方法可以做到这一点，还是我只需要使用流式传输然后发回响应？

原型文件：


service SimpleService {
rpc SimpleRPC (stream SimpleData) returns (stream SimpleData) {}
}

message SimpleData {
string msg = 1;
}

转码：


client := pb.NewSimpleServiceClient(conn)
stream, err := client.SimpleRPC(context.Background())
waitc := make(chan struct{})

msg := &pb.SimpleData{"sup"}
go func() {
for {
   stream.Send(msg)
    }
}()
<-waitc
stream.CloseSend()

【问题讨论】：

它基于 tcp，发送成功的消息就足够了。但否则为每条发送的消息添加一个 tx id。将其传递回响应中，以便发件人可以匹配查询以获得响应。
所以发回带有消息ID或其他内容的异步响应，如果没有响应则重新发送？
可能是异步的，也可能不是。这取决于协议。但是，同样，这是基于 tcp 的流，所以发送/接收是可靠的。
不需要，TCP 连接会自动进行数据包级别的确认，为此添加另一层只是矫枉过正。正如@mh-cbon 之前所说，它是可靠的。
但是如果我在服务器离线的那一刻发送消息 if stream.Send(msg) gRCP 会告诉我有错误吗？

标签： go streaming grpc bidirectional

【解决方案1】：

不幸的是，gRPC 不能很好地处理这种情况。它也不处理单向流式传输，甚至处理单个 RPC 请求。故障模式（没有流式的简单情况）是这样的：

gRPC 客户端向服务器发送请求，超时在T
服务器成功响应客户端T-1sec
将消息发送给客户端和/或解析它需要2sec，将其放在T+1sec（超过超时期限）
客户端在时间 T 之前的请求超时，并看到错误

最好的解决方案是找出一种方法来避免这样做，从服务器的角度使协议无状态 - 换句话说，如果客户端看到失败，让他们重试。或者，您可以反转连接的方向，以便“服务器”向客户端发送 RPC，客户端以 ack 响应。

如果服务器存储的内容取决于客户端的状态，则很难做到这一点。例如，假设服务器有一个有序的数据队列，它试图发送给客户端，并且它需要知道客户端在从队列中删除项目之前看到的最后一件事。这里的选项是：

如果你有一个 RPC，把它变成一个（非常短暂的）双向 RPC 流，客户端可以发送一个 ClientRequest 对象，内部是 oneof 一个 ClientActualRequest 对象或一个空的 @987654329 @ 表示客户端看到消息的对象。然后，在客户端的实现中，它首先发送一个ClientRequest { ClientActualRequest {...} }，然后在接收到来自服务器的响应后发送一个ClientRequest { ClientSuccess {} }，然后终止流。（当然，服务器也必须知道如何处理。）
如果您在其中一个/两个方向上进行流式传输，则会变得更加困难。在最简单的情况下，您可以执行与上述单个 RPC 情况相同的事情，其中首先客户端发送一个真实请求，然后在来自服务器的每次更新时发送一个确认。重要的是，客户端必须按顺序处理流，服务器必须按顺序处理确认，否则服务器可能认为客户端已经看到了所有消息，而它只看到了一个子集。（请注意，gRPC 保证按顺序交付，因此只有在您的客户端实现中重新排序才会导致问题；可能只有在您使用异步 gRPC API 时才有可能，除非您正在做一些非常奇怪的事情。）对此的另一个转折将是，您可以让客户端将last_known_sequence_number 作为协议的一部分发送到服务器，而不是发送确认消息，并且服务器只能安全地删除直到该序列号的项目；但是，如果您这样做，则没有简单的方法可以将更复杂的状态传回，因为它只是一个序列号。
如果在客户端上按顺序处理不可行，那么您必须完全发疯。现在您的协议需要服务器将序列号放在队列中的每条消息上，并且您的客户端的ClientSuccess 对象需要对您确认的序列号进行编码，以便服务器知道它可以删除哪些内容。这基本上是 TCP 在内部所做的，但当然只是在从服务器到客户端的方向上进行确认......显然，如果您将您的解决方案与 TCP 进行比较，我们正在谈论一些非常复杂的东西。

请记住，在所有这些情况下，即使 RPC 成功，也不能保证客户端可以看到服务器的 ack，在我们的新协议中，不能保证客户端可以看到客户端的 ack服务器。出于这个原因，您的客户端需要能够在连接重置时处理看到重复的请求（最多您允许它一次拥有的任何数量的飞行请求）。除非你实现一个“ack of an ack”，否则就是：-P。

【讨论】：