Golang gRPC ServerStream 报错：错误传输：流已完成或 WriteHeader 已被调用

在IT行业这个发展更新速度很快的行业，只有不停止的学习，才不会被行业所淘汰。如果你是Golang学习者，那么本文《Golang gRPC ServerStream 报错：错误传输：流已完成或 WriteHeader 已被调用》就很适合你！本篇内容主要包括##content_title##，希望对大家的知识积累有所帮助，助力实战开发！

我的 grpc 服务器流函数出现了一个有趣的错误，我快抓狂了。 无法通过阅读 grpc godoc 或在线其他地方找到任何可能的原因。 希望更熟悉 go 和 grpc 流的人能够为我指明正确的方向。

我的实现尝试遵循 grpc.io 网站上的基本服务器流示例。

有问题的 protobuf 定义：

service arraybasedcache {
  ...
  rpc getrecord (getrecordrequest) returns (stream messageresponse) {}
}

message getrecordrequest {
  string key = 1;
}

message messageresponse {
  string message = 1;
}

grpc 服务器处理程序：

func (ctlr *cacheclientcontroller) getrecord(req *svcgrpc.getrecordrequest, stream svcgrpc.arraybasedcache_getrecordserver) error {
    key := req.getkey()
    if ctlr.inputchannels[key] == nil {
        return errors.new("requested record has expired")
    }
    msgs, e1 := ctlr.client.readarrayrecord(key)
    if e1 != nil {
        panic(e1)
    }
    log.printf("messages: %v", msgs)
    for i := 0; i < len(msgs); i++ {
        log.printf("trying to write message: %v", msgs[i])
        if e2 := stream.send(&svcgrpc.messageresponse{message: msgs[i]}); e2 != nil {
            log.printf("writing message %d of %d to stream failed", i+1, len(msgs))
            panic(e2)
        }
    }
    return nil
}

我的 grpc 客户端实现：

func (s *grpcservice) getrecord(key string) (svcgrpc.arraybasedcache_getrecordclient, error) {
    req := &svcgrpc.getrecordrequest{key: key}
    ctx, cancelfunc := context.withtimeout(context.background(), defaulttimeout)
    defer cancelfunc()
    resp, err := s.grpcclient.getrecord(ctx, req)
    if err != nil {
        return nil, err
    }
    return resp, nil
}

func (s *grpcservice) streamtoarray(stream svcgrpc.arraybasedcache_getrecordclient) ([]string, error) {
    out := []string{}
    for {
        msg, err := stream.recv()
        if err == io.eof {
            break
        }
        if err != nil {
            return nil, errors.new("unexpected error reading stream")
        }
        out = append(out, msg.message)
    }
    return out, nil
}

我的测试客户端中的入口点

stream, err := c.getrecord(testkey)
    if err != nil {
        log.printf("retrieving record stream %v failed", testkey)
    }
    arr, err := c.streamtoarray(stream)
    if err != nil {
        log.printf("failed to unmarshall message stream to array")
    }
    log.printf("retrieved messages %v", arr)

遇到的错误：

2020/10/26 22:43:36 Messages: [abc def hij]
2020/10/26 22:43:36 trying to write message: abc
2020/10/26 22:43:36 Writing message 1 of 3 to stream failed
panic: rpc error: code = Internal desc = transport: transport: the stream is done or WriteHeader was already called

goroutine 82 [running]:
github.com/TasSM/app/service/controller.(*cacheClientController).GetRecord(0xc0001e4cc0, 0xc000176fc0, 0xd6ef80, 0xc00022e5f0, 0xc0001e4cc0, 0xc00039fa00)
        C:/Users/myuser/Documents/repos/app/src/service/controller/controller.go:96 +0x5ad
github.com/TasSM/app/service/svcgrpc._ArrayBasedCache_GetRecord_Handler(0xc11a80, 0xc0001e4cc0, 0xd6ca00, 0xc0001dc300, 0x11dec30, 0xc000392000)
        C:/Users/myuser/Documents/repos/app/src/service/svcgrpc/cacheservice_grpc.pb.go:192 +0x13d
google.golang.org/grpc.(*Server).processStreamingRPC(0xc0001a7a40, 0xd70300, 0xc000108c00, 0xc000392000, 0xc0001e4de0, 0x119fb00, 0x0, 0x0, 0x0)
        C:/Users/myuser/go/pkg/mod/google.golang.org/[email protected]/server.go:1457 +0x15cb
google.golang.org/grpc.(*Server).handleStream(0xc0001a7a40, 0xd70300, 0xc000108c00, 0xc000392000, 0x0)
        C:/Users/myuser/go/pkg/mod/google.golang.org/[email protected]/server.go:1537 +0x1309
google.golang.org/grpc.(*Server).serveStreams.func1.2(0xc000123140, 0xc0001a7a40, 0xd70300, 0xc000108c00, 0xc000392000)
        C:/Users/myuser/go/pkg/mod/google.golang.org/[email protected]/server.go:871 +0xe0
created by google.golang.org/grpc.(*Server).serveStreams.func1
        C:/Users/myuser/go/pkg/mod/google.golang.org/[email protected]/server.go:869 +0x349
exit status 2

解决方案

我认为问题出在 getrecord grpc 客户端实现中，特别是您正在使用的上下文实现中。

通过使用 context.withtimeout 并在同一方法中调用 defer cancelfunc()，您基本上可以在从 getrecord 方法返回之前关闭流。

如果您想仍然使用 context.withtimeout 实现，请不要在 getrecord 方法中使用 cancelfunc，而是从中返回 cancelfunc，或者将 ctx 传递给 get记录方法。

@ eminlala 确定了流终止的根本原因。

由于流意味着长时间运行 - 实时返回结果 - 通常您不希望对流式结果设置截止日期。

如果您想对建立流连接的 dial 操作设置超时，您可以通过 dialoption WithTimeout 执行此操作：

grpc.dial(addr, grpc.withtimeout(defaulttimeout))

或使用dialcontext：

ctx, cancelFunc := context.WithTimeout(context.Background(), defaultTimeout)
defer cancelFunc()

grpc.DialContext(ctx, addr)

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~