转到 gRPC 流范围

Question

TLDR；如果我的 Go gRPC 双向流服务器在 Recv/Send 上无限循环，当我的客户端 stream 超出范围并被 GC 处理时，我该如何关闭连接？

假设我正在使用 Go 进行 gRPC 双向流式调用：

syntax = "proto3";

package mystical;

service Unicorn {
    rpc RainbowStream(stream Rainbow) returns (stream Rainbow);
}

我有一个服务器实现，我将在我的 gRPC 服务器上注册它：

type Server struct {}

func (serv *Server) RainbowStream(stream proto.Unicorn_RainbowStreamServer) error {
    // Stub implementation
    // Just sends the rainbows back
    for {
        msg, err := stream.Recv()
        if err != nil {
            return err
        }
        if err := stream.Send(msg); err != nil {
            return err
        }
    }
}

然后我有一个客户端，省略了一些 gRPC 细节：

type Client struct {
    proto.UnicornClient
}

func (c *Client) TastyColors() (stream proto.Unicorn_RainbowStreamClient, err error) {
    return c.RainbowStream()
}

我这样做：

func somewhereDeepInMyCode() {

    stream, err := aClient.TastyColors()
    // ...

    go func() {
       stream.Send(msg)
       // ...
       // eventually this goroutine ends
       // and `stream` would get GC'd
    }
}

当 stream 超出范围时，gRPC 流调用会自行清理吗？通常，据我了解，服务器必须在其 RainbowStream 调用中 return 关闭连接。如果它不能自行清理，我该如何实现？

Answer 1

使用取消 (https://golang.org/pkg/context/#WithCancel) 上下文从客户端取消：

func somewhereDeepInMyCode() {
    ctx, cancel := context.WithCancel(context.Background())
    stream, err := aClient.TastyColors(ctx)
    // ...

    go func() {
        defer cancel()
        stream.Send(msg)
        // ...
        // eventually this goroutine ends
        // and `stream` would get GC'd
    }
}

服务器可以根据需要处理客户端取消以进行清理：

for {
    select {
    ...
    case <-stream.Context().Done():
    // cleanup
    ...
    }
}

转到 gRPC 流范围

Go gRPC Stream Scope

go

grpc