go - 如何关闭服务器的grpc流

Question

在grpc中，客户端可以调用CloseSend来关闭到服务器的流，但服务器似乎无法切断与客户端的连接。

Answer 1

解决这个问题的最好方法是使用三个 goroutines。

问题在于 bidi 处理程序将被阻塞在 Recv() 中，因此当它想要完成流时无法轻易“返回到调用者”。

另一个问题是，如果网络条件不好，或者另一端不合作，Send() 可能会阻塞(这是您首先要终止连接的一个非常常见的原因!)

通常，您还希望从接收方循环外部向发送方发送消息——使用双向流的典型情况是与世界其他地方进行实际通信。对于更隔离的请求/响应模式，其他模式通常更易于使用。允许同步发送/接收循环之外的人对消息进行排队会在混合中引入额外的同步。

因此，您可以在 Bidi 处理程序中构建您的代码:

1. 创建一些可等待的原语——一个空的 chan、一个 WaitGroup，或者只是一个带有取消的上下文
2. 启动一个在循环中调用 Recv() 的 goroutine，并分派(dispatch)传入的请求
3. 为外发消息创建一个 channel ，具有一定的最大深度
4. 启动另一个 goroutine 从传出的 chan 中出队，并调用 Send()
5. 在等待原语上阻止传入的 Bidi 请求处理程序

现在，当您想要尝试发送给发件人 channel 时，请在 select{} 中执行此操作，如果传出 channel 已满，它将转到 default。您可以将其用作另一端应断开连接的信号，或者根据需要丢弃该消息以避免无限缓冲。

当想要断开客户端与服务器端的连接时，您应该解除阻塞 bidi 处理程序阻塞的可等待原语。这最终会使 Send() 和 Recv() 函数返回错误，之后您可以安全地声明连接“完成”。

还有远程客户端关闭的常规情况，并且 Recv() 调用返回错误，在这种情况下，您还需要将其关闭。

如果你想真正关闭所涉及的 channel ，并且除了断开无响应客户端的连接之外还支持上下文取消，你最终将不得不将 channel 与额外的互斥体同步，以避免意外地发送一个关闭的 channel ——没有以其他方式安全地执行此操作的方法，因为 Send()/Recv() 函数在 bidi 处理程序返回之前不会解除阻塞。 (如果没有外部发送者，发送者的所有排队都发生在接收 goroutine 中，这不是问题。不过这只适用于非常简单的服务器。)

有些伪代码:

func (m *MyThing) MyBidiServer(stream somepb.Thing_ThingServer) {
    can := make(chan struct{})
    thingRunner := &ThingRunner{
        cancel: can,
        send: make(chan *somepb.OutgoingMessage, 100),
    }
    m.addRunner(thingRunner) // assume this is mutexed
    defer m.removeRunner(thingRunner)
    go thingRunner.recvLoop(stream)
    go thingRunner.sendLoop(stream)
    // could be select here if you also have a context
    <-can
}

func (t *ThingRunner) recvLoop(stream somepb.Thing_ThingServer) {
    for {
        msg, err := stream.Recv()
        if err != nil {
            break
        }
    }
    t.cancelSafely() // in case client disconnected
}

func (t *ThingRunner) sendLoop(stream somepb.Thing_ThingServer) {
    for msg := range t.send {
        if err := stream.Send(msg); err != nil {
            break
        }
    }
    t.cancelSafely() // in case of network error
}

func (t *ThingRunner) SendMessage(ctx context.Context, msg *somepb.Message) error {
    if t.isCanceled() {
        return ThingClosedError
    }
    select {
        case t.send <- msg:
            // all is well
        case <-ctx.Done():
            return ctx.Err()
        default:
            // the client is too slow -- disconnect it
            t.cancelSafely()
            return ClientTooSlowError
    }
    return nil
}

您还需要注意一些小的管理细节，但这就是要点。 cancelSafely() 需要取消阻塞 cancel chan，但它也需要不导致 future 对 t.send 的写入发生 panic ，因为在关闭的 chan 上发送将无条件地发生 panic 。