gRPC 四种流式通信模型：Unary、Server/Client Streaming 与 Bidirectional 实战

syntax = "proto3";

package chat;

service ChatService {
    // 1. Unary：一元调用，请求-响应各一个
    rpc SendMessage (Message) returns (MessageReply);

    // 2. Server Streaming：客户端发一个，服务端回一堆（推送/订阅）
    rpc Subscribe (SubscribeReq) returns (stream Event);

    // 3. Client Streaming：客户端发一堆，服务端回一个（上传/聚合）
    rpc Upload (stream Chunk) returns (UploadReply);

    // 4. Bidirectional Streaming：两端都流式，全双工
    rpc Chat (stream ChatMsg) returns (stream ChatMsg);
}

message Message { string text = 1; }
message MessageReply { string reply = 1; }
message SubscribeReq { string topic = 1; }
message Event { bytes data = 1; }
message Chunk { bytes payload = 1; }
message UploadReply { int64 size = 1; }
message ChatMsg { string sender = 1; string text = 2; }

// 服务端：订阅推送（stream）
func (s *chatServer) Subscribe(req *SubscribeReq, stream chat.ChatService_SubscribeServer) error {
    events := make(chan *Event)   // 假设业务从这里拿到事件
    defer close(events)

    go s.produceEvents(req.Topic, events)

    for ev := range events {
        // 每次 Send 发送一个流消息
        if err := stream.Send(ev); err != nil {
            return err            // 客户端断开/网络异常
        }
    }
    return nil
}

func subscribe() error {
    conn, _ := grpc.NewClient("localhost:50051", grpc.WithTransportCredentials(...))
    defer conn.Close()
    client := chat.NewChatServiceClient(conn)

    stream, err := client.Subscribe(ctx, &SubscribeReq{Topic: "stock.tsla"})
    if err != nil { return err }

    for {
        ev, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            break                      // 服务端正常结束
        }
        if err != nil { return err }   // 异常断开
        handleEvent(ev)
    }
    return nil
}

// 服务端
func (s *ClientServer) Upload(stream chat.ChatService_UploadServer) error {
    var total int64
    for {
        chunk, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            break                     // 客户端发完
        }
        if err != nil { return err }
        total += int64(len(chunk.Payload))
        // 逐块写入存储……
    }
    return stream.SendAndClose(&UploadReply{Size: total})
}

// 客户端
stream, _ := client.Upload(ctx)
// 循环发
for _, chunk := range fileChunks(file) {
    if err := stream.Send(&Chunk{Payload: chunk}); err != nil {
        return err
    }
}
// 发完关闭发送端，等待服务端回执
reply, err := stream.CloseAndRecv()
_ = reply.Size

// 服务端：双向流
func (s *ChatServer) Chat(stream chat.ChatService_ChatServer) error {
    for {
        in, err := stream.Recv()      // 收一条客户端消息
        if errors.Is(err, io.EOF) {
            return nil
        }
        if err != nil { return err }

        // 处理并回复（可并发 Send）
        if err := stream.Send(&ChatMsg{
            Sender: "server",
            Text:  "收到: " + in.Text,
        }); err != nil {
            return err
        }
    }
}

关键点 1：Send 与 Recv 由同一 stream 对象完成，但底下的 Session 是 HTTP/2 全双工。

关键点 2：Recv 和 Send 可以各自在一个 goroutine 里独立跑，互不阻塞。

关键点 3：要控制流控背压——如果服务端 Send 太快、客户端 Recv 太慢，
  底层 TCP 缓冲会积压，最终触发阻塞，需要配合并发限制/令牌桶。

典型模式：
  - 消息 → worker 池（限并发）
  - worker 结果 → channel
  - 单独发送 goroutine 从 channel 取出 Send

// 服务端一元拦截器：统一鉴权 / 日志 / 限流
func unaryAuthInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok { return nil, status.Error(codes.Unauthenticated, "no metadata") }
    if token := md.Get("authorization"); len(token) == 0 {
        return nil, status.Error(codes.Unauthenticated, "missing token")
    }
    // 鉴权通过，进入实际 handler
    return handler(ctx, req)
}

// 注册
grpc.NewServer(grpc.UnaryInterceptor(unaryAuthInterceptor))

// 返回结构化错误码（客户端能精确处理）
return nil, status.Error(codes.NotFound, "user id not found")

// 客户端侧判断
if st, ok := status.FromError(err); ok {
    switch st.Code() {
    case codes.NotFound:
        // 未找到逻辑
    case codes.Unavailable:
        retry()   // 可重试
    }
}

// gRPC 客户端默认带自动重连（当 resolver 配置时）
// 但流式连接中断后，需要手动重建 stream：
for {
    stream, err := newChatStream(ctx)
    if err != nil {
        time.Sleep(backoff())   // 指数退避
        continue
    }
    err = consumeStream(stream) // 阻塞，断开时返回错误
    // 记录重连次数、保留进度、指数退避
}

□ Protobuf 字段变更是非破坏的，但别复用字段号
□ 流式接口一定设超时/心跳，防 TCP 半开连接
□ 别把大对象一次性放内存，改用流式传
□ 记住 Recv 返回 io.EOF 才是正常结束
□ 用 load-balancer / resolver 做到连接级复用
□ 服务端 Send 阻塞时用并发 worker 避免卡死
□ 监控流式打开时间长度和消息速率，防泄漏