GRPC：保留对流的引用，以将数据发送到多个客户端

Question

我从使用go的GRPC开始。我阅读了官方文档，并举了几个例子。

在大多数示例中，您不识别客户端，而是使用流读取/写入数据。我看到上下文中有用于检索身份验证信息的应用编程接口，并且可以为聊天请求识别客户端。但是，如果我想根据客户端标识保留流的引用/索引呢？

例如，

假设我在聊天室有3个用户。我将rpc表示为（也可能是服务器流）

rpc Chat(stream ChatRequest) returns (stream ChatResponse) {}

例如，一个用户向该组发送一条消息，该组需要向另两个用户发送。因此，如果我需要通过当前为这些用户打开的流发送它，那么保存流的引用有多安全。

实施将像...

type chatServiceServer struct {
    // keep a map of subscribers / users currently connected; protect with mutex
}

func (s *chatServiceServer) Chat(stream pb.ChatService_ChatServer) error {
    // md, ok := metadata.FromIncomingContext(stream.Context())
    // p, ok := peer.FromContext(ctx)
    // ... identify client from above

    for {
        // save the message to DB
        // find other users in the chatroom is currently connected
        // if so, stream.Send(m)
        // else notify ....
    }
}

但是，我看到API文档中的警告，想知道更好的方法。

https://godoc.org/google.golang.org/grpc#ServerStream
// It is safe to have a goroutine calling SendMsg and another goroutine
// calling RecvMsg on the same stream at the same time, but it is not safe
// to call SendMsg on the same stream in different goroutines.
SendMsg(m interface{}) error

任何订阅（事件等）都会出现类似的用例，并且需要根据客户id进行通知。任何示例代码、文章都会很好。

非常感谢。

Answer 1

将流保存到其他地方使用应该是安全的，但是您只能在单个goroutine中调用SendMsg（对于RecvMsg来说，同样的限制是独立的）。所以这意味着如果您在方法处理程序中这样做：

for {
  if err := stream.Recv(req); err != nil {
    return err
  }
  for _, s := range allStreams[req.ID] {
    if err := s.Send(req.Message); err != nil { /* remove s from allStreams */ }
  }
}

然后对s.Send的调用必须由锁保护，因为这些处理程序中有多个可能同时运行。（另外，allStreams被假定为一个map[ID][]流，在这种情况下，它还必须由锁保护。）