客户端断开连接时如何正确使用 ctx.Done()？答案

【问题标题】：How to properly use ctx.Done() when client is disconnected?客户端断开连接时如何正确使用 ctx.Done()？
【发布时间】：2018-04-01 06:12:12
【问题描述】：

如果客户端将因网络错误而断开连接，在我的情况下，服务器必须关闭 pub/sub 连接。我知道ctx.Done() 功能，但不知道如何在我的情况下正确使用它。谁能解释一下？

grpc-go：1.7.0

转到版本 go1.8.4

func (a *API) Notifications(in *empty.Empty, stream pb.Service_NotificationsServer) error {
    ctx := stream.Context()
    _, ok := user.FromContext(ctx)
    if !ok {
        return grpc.Errorf(codes.Unauthenticated, "user not found")
    }

    pubsub := a.redisClient.Subscribe("notifications")
    defer pubsub.Close()

    for {
        msg, err := pubsub.ReceiveMessage()
        if err != nil {
            grpclog.Warningf("Notifications: pubsub error: %v", err)
            return grpc.Errorf(codes.Internal, "pubsub error %v", err)
        }

        notification := &pb.Notification{}
        err = json.Unmarshal([]byte(msg.Payload), notification)
        if err != nil {
            grpclog.Warningf("Notifications: parse error: %v", err)
            continue
        }
        if err := stream.Send(notification); err != nil {
            grpclog.Warningf("Notifications: %v", err)
            return err
        }
        grpclog.Infof("Notifications: send msg %v", notification)
    }
}

【问题讨论】：

标签： go grpc channel

【解决方案1】：

您应该从调用函数（或可以访问上下文的任何地方）取消上下文，并在 Done() 上执行适当的操作，并签入 select 语句。

Done 用于在 select 语句中使用

当代表此上下文完成的工作应该被取消时，Done 返回一个关闭的通道。如果这个上下文永远不能被取消，Done 可能会返回 nil。对 Done 的连续调用返回相同的值。

和

WithCancel 返回具有新 Done 通道的 parent 副本。返回的上下文的 Done 通道在调用返回的取消函数或父上下文的 Done 通道关闭时关闭，以先发生者为准。

取消此上下文会释放与其关联的资源，因此代码应在此上下文中运行的操作完成后立即调用取消。

go func() {
    for {
        select {
        case <-ctx.Done():
            return // returning not to leak the goroutine
        case dst <- n:
            n++
        }
    }
}()

【讨论】：

【解决方案2】：

您可以使用select。不是从函数中正常获取数据，而是使用通道获取数据并使用 go 例程来处理它。像这样的东西：

func (a *API) Notifications(in *empty.Empty, stream 
    pb.Service_NotificationsServer) error {
    ctx := stream.Context()
    _, ok := user.FromContext(ctx)
    if !ok {
        return grpc.Errorf(codes.Unauthenticated, "user not found")
    }

    pubsub := a.redisClient.Subscribe("notifications")
    defer pubsub.Close()

    // I can not build the code, so I assume the msg in your code Message struct
    c := make(chan Message)
    go func() {
        for {
            msg, err := pubsub.ReceiveMessage()
            if err != nil {
                grpclog.Warningf("Notifications: pubsub error: %v", err)
                close(c)
                return grpc.Errorf(codes.Internal, "pubsub error %v", err)
            }
            c<- msg
        }
    }()

    for {
        select {
            case msg, ok  := <-c:
                if !ok {
                    // channel is closed handle it
                }
                notification := &pb.Notification{}
                err = json.Unmarshal([]byte(msg.Payload), notification)
                if err != nil {
                    grpclog.Warningf("Notifications: parse error: %v", err)
                    continue
                }
                if err := stream.Send(notification); err != nil {
                    grpclog.Warningf("Notifications: %v", err)
                    return err
                }
                grpclog.Infof("Notifications: send msg %v", notification)
            case <- ctx.Done():
                // do exit logic. some how close the pubsub, so next 
                // ReceiveMessage() return an error
                // if forget to do that the go routine runs for ever 
                // until the end of main(), which I think its not what you wanted
                pubsub.Close() // Its just pseudo code
                return
        }
    }
}

从频道读取消息（我假设类型是消息），并使用select 的力量。

这个场景中另外两个相关的东西：

确保 go 例程在完成此函数后结束。我无法猜测，因为我不知道代码，但我假设有一个 Close() 方法用于关闭 pubsub 所以下一个 ReceiveMessage 返回错误。（我看到延迟完成了我希望的工作）
如果ReceiveMessage 在ctx.Done 之前出现错误，您可以关闭通道然后中断循环。

【讨论】：