如何使用gracefulStop关闭所有grpc服务器流？

你在学习Golang相关的知识吗？本文《如何使用gracefulStop关闭所有grpc服务器流？》，主要介绍的内容就涉及到，如果你想提升自己的开发能力，就不要错过这篇文章，大家要知道编程理论基础和实战操作都是不可或缺的哦！

我正在尝试停止从服务器端连接到流服务器的所有客户端。 实际上我正在使用 gracefulstop 方法来优雅地处理它。

我正在通道上等待 os.interrupt 信号来执行 grpc 的正常停止。但当客户端连接时，它会卡在 server.gracefulstop() 上。

func (s *Service) Subscribe(_ *empty.Empty, srv clientapi.ClientApi_SubscribeServer) error {
    ctx := srv.Context()

    updateCh := make(chan *clientapi.Update, 100)
    stopCh := make(chan bool)
    defer func() {
        stopCh<-true
        close(updateCh)
    }

    go func() {
        ticker := time.NewTicker(1 * time.Second)
        defer func() {
            ticker.Stop()
            close(stopCh)
        }
        for {
            select {
            case <-stopCh:
                return
            case <-ticker.C:
                updateCh<- &clientapi.Update{Name: "notification": Payload: "sample notification every 1 second"}
            }
        }
    }()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()

        case notif := <-updateCh:
            err := srv.Send(notif)
            if err == io.EOF {
                return nil
            }

            if err != nil {
                s.logger.Named("Subscribe").Error("error", zap.Error(err))
                continue
            }
        }
    }
}

我期望方法 ctx.done() 中的 context 可以处理它并打破 for 循环。 如何关闭像这样的所有响应流？

解决方案

为您的 grpc 服务创建一个全局 context。因此，浏览各个部分：

• 每个 grpc 服务请求都将使用此上下文（以及客户端上下文）来满足该请求
• os.interrupt 处理程序将取消全局上下文；从而取消任何当前正在运行的请求
• 最后发出 server.gracefulstop() - 应该等待所有活动的 grpc 调用完成（如果它们没有立即看到取消）

例如，在设置 grpc 服务时：

pctx := context.background()
globalctx, globalcancel := context.withcancel(pctx)

mysrv := myservice{
    gctx: globalctx
}

s := grpc.newserver()
pb.registermyservice(s, mysrv)

os.interrupt 处理程序启动并等待关闭：

globalcancel()
server.gracefulstop()

grpc 方法：

func(s *myservice) somerpcmethod(ctx context.context, req *pb.request) error {

    // merge client and server contexts into one `mctx`
    // (client context will cancel if client disconnects)
    // (server context will cancel if service ctrl-c'ed)

    mctx, mcancel := mergecontext(ctx, s.gctx)

    defer mcancel() // so we don't leak, if neither client or server context cancels

    // rpc work goes here
    // rpc work goes here
    // rpc work goes here

    // pass mctx to any blocking calls:
    // - http rest calls
    // - sql queries etc.
    // - or if running a long loop; status check the context occasionally like so:

    // example long request (10s)
    for i:=0; i<10*1000; i++ {
        time.sleep(1*time.milliscond)

        // poll merged context
        select {
            case <-mctx.done():
                return fmt.errorf("request canceled: %s", mctx.err())
            default:
        }
    }
}

还有：

func mergeContext(a, b context.Context) (context.Context, context.CancelFunc) {
    mctx, mcancel := context.WithCancel(a) // will cancel if `a` cancels

    go func() {
        select {
        case <-mctx.Done(): // don't leak go-routine on clean gRPC run
        case <-b.Done():
            mcancel() // b canceled, so cancel mctx 
        }
    }()

    return mctx, mcancel
}

通常，客户端需要假设 rpc 可以随时终止（例如，由于连接错误或服务器电源故障）。所以我们要做的是 gracefulstop，休眠一小段时间，让正在进行的 rpc 有机会自然完成，然后硬 stop 服务器。如果您确实需要使用此终止信号来结束 rpc，那么 @colminator 的答案可能是最佳选择。但这种情况应该不常见，如果您确实发现有必要在服务器关闭时手动结束流式 rpc，您可能需要花一些时间分析您的设计。

理论要掌握，实操不能落！以上关于《如何使用gracefulStop关闭所有grpc服务器流？》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！