4、gRPC通信模式：一元RPC、服务端流式RPC、客户端流式RPC、双向流式RPC，适用场景与代码示例

gRPC 基于 HTTP/2 协议，提供了四种核心通信模式。理解这四种模式的区别与适用场景，是设计高性能微服务接口的基础。本章将逐一剖析每种模式的工作原理、适用场景，并给出基于 Protocol Buffers 的完整代码示例。

4.1 四种通信模式总览

下表从调用方式、数据流方向、典型延迟和适用场景四个维度，对四种模式进行横向对比：

通信模式 调用方式 数据流方向 典型延迟 适用场景
一元 RPC 请求-响应 客户端 → 服务端 → 客户端 低 传统 CRUD、查询、认证
服务端流式 RPC 请求-流式响应 客户端 → 服务端 → 客户端（多次） 中 大文件下载、日志推送、实时监控
客户端流式 RPC 流式请求-响应 客户端（多次）→ 服务端 → 客户端 中 批量数据上传、传感器数据聚合
双向流式 RPC 流式请求-流式响应 客户端 ↔ 服务端（独立双向） 高 实时聊天、游戏同步、协同编辑
关键认知： 四种模式并非互斥。在复杂业务中，一个微服务接口可以同时暴露多种模式，例如：查询接口用一元 RPC，批量导出用服务端流式 RPC，数据采集用客户端流式 RPC。

4.2 一元 RPC（Unary RPC）

一元 RPC 是最基础的通信模式，客户端发送一个请求，服务端返回一个响应。其行为与传统的 HTTP REST 接口完全一致，但底层使用 HTTP/2 和 Protobuf 序列化，性能更高。

适用场景

  • 用户登录认证（发送凭证，返回 Token）
  • 单条数据查询（根据 ID 获取订单详情）
  • 配置下发（客户端请求，服务端返回最新配置）

Proto 定义

syntax = "proto3";

service UserService {
  // 一元 RPC：获取用户信息
  rpc GetUser (GetUserRequest) returns (GetUserResponse);
}

message GetUserRequest {
  int64 user_id = 1;
}

message GetUserResponse {
  int64 user_id = 1;
  string name = 2;
  string email = 3;
}

服务端实现（Go 示例）

func (s *userServer) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.GetUserResponse, error) {
    // 模拟数据库查询
    user := &pb.GetUserResponse{
        UserId: req.UserId,
        Name:   "张三",
        Email:  "zhangsan@example.com",
    }
    return user, nil
}

客户端调用（Go 示例）

resp, err := client.GetUser(ctx, &pb.GetUserRequest{UserId: 1001})
if err != nil {
    log.Fatalf("调用失败: %v", err)
}
log.Printf("用户信息: %+v", resp)

4.3 服务端流式 RPC（Server Streaming RPC）

客户端发送一个请求，服务端以流的形式返回多个响应消息。客户端通过迭代器逐条读取，直到流结束。这种模式非常适合服务端主动推送大量数据的场景。

适用场景

  • 日志实时推送（客户端订阅，服务端持续推送日志行）
  • 大文件分块下载（服务端将文件切分为多个 chunk 发送）
  • 监控指标流（服务端持续上报 CPU、内存等指标）

Proto 定义

service LogService {
  // 服务端流式 RPC：订阅日志流
  rpc SubscribeLogs (SubscribeLogsRequest) returns (stream LogEntry);
}

message SubscribeLogsRequest {
  string log_level = 1; // 日志级别：INFO / WARN / ERROR
}

message LogEntry {
  int64 timestamp = 1;
  string level = 2;
  string message = 3;
}

服务端实现（Go 示例）

func (s *logServer) SubscribeLogs(req *pb.SubscribeLogsRequest, stream pb.LogService_SubscribeLogsServer) error {
    // 模拟持续推送日志
    for i := 0; i < 100; i++ {
        entry := &pb.LogEntry{
            Timestamp: time.Now().Unix(),
            Level:     req.LogLevel,
            Message:   fmt.Sprintf("日志条目 #%d", i),
        }
        if err := stream.Send(entry); err != nil {
            return err
        }
        time.Sleep(500 * time.Millisecond) // 模拟生产间隔
    }
    return nil
}

客户端调用（Go 示例）

stream, err := client.SubscribeLogs(ctx, &pb.SubscribeLogsRequest{LogLevel: "INFO"})
if err != nil {
    log.Fatalf("订阅失败: %v", err)
}
for {
    entry, err := stream.Recv()
    if err == io.EOF {
        break // 流结束
    }
    if err != nil {
        log.Fatalf("读取流失败: %v", err)
    }
    log.Printf("收到日志: %+v", entry)
}

4.4 客户端流式 RPC（Client Streaming RPC）

客户端以流的形式发送多个请求消息，服务端在接收完所有消息后，返回一个统一的响应。这种模式适用于客户端需要批量提交数据的场景。

适用场景

  • 批量数据上传（上传 10 万条用户记录，服务端返回处理结果）
  • 传感器数据聚合（多个传感器持续上报数据，服务端汇总后返回统计值）
  • 大文件上传（客户端将文件分块上传，服务端最后返回文件 URL）

Proto 定义

service SensorService {
  // 客户端流式 RPC：上报传感器数据
  rpc ReportSensorData (stream SensorReading) returns (AggregationResult);
}

message SensorReading {
  string sensor_id = 1;
  double temperature = 2;
  double humidity = 3;
  int64 timestamp = 4;
}

message AggregationResult {
  int32 total_readings = 1;
  double avg_temperature = 2;
  double avg_humidity = 3;
}

服务端实现（Go 示例）

func (s *sensorServer) ReportSensorData(stream pb.SensorService_ReportSensorDataServer) error {
    var total int32
    var sumTemp, sumHum float64

    for {
        reading, err := stream.Recv()
        if err == io.EOF {
            // 客户端发送完毕，返回聚合结果
            result := &pb.AggregationResult{
                TotalReadings:   total,
                AvgTemperature:  sumTemp / float64(total),
                AvgHumidity:     sumHum / float64(total),
            }
            return stream.SendAndClose(result)
        }
        if err != nil {
            return err
        }
        total++
        sumTemp += reading.Temperature
        sumHum += reading.Humidity
    }
}

客户端调用（Go 示例）

stream, err := client.ReportSensorData(ctx)
if err != nil {
    log.Fatalf("创建流失败: %v", err)
}

// 模拟发送 10 条传感器数据
for i := 0; i < 10; i++ {
    reading := &pb.SensorReading{
        SensorId:    "sensor-001",
        Temperature: 25.0 + float64(i)*0.5,
        Humidity:    60.0 + float64(i)*0.3,
        Timestamp:   time.Now().Unix(),
    }
    if err := stream.Send(reading); err != nil {
        log.Fatalf("发送失败: %v", err)
    }
    time.Sleep(200 * time.Millisecond)
}

// 关闭发送端并接收响应
result, err := stream.CloseAndRecv()
if err != nil {
    log.Fatalf("接收结果失败: %v", err)
}
log.Printf("聚合结果: %+v", result)

4.5 双向流式 RPC（Bidirectional Streaming RPC）

客户端和服务端可以独立地、并发地发送和接收消息流。双方都不需要等待对方完成，可以实现真正的全双工通信。这是 gRPC 最强大的通信模式。

适用场景

  • 实时聊天系统（用户 A 和用户 B 通过服务端中转消息）
  • 多人游戏状态同步（客户端上报操作，服务端广播其他玩家状态）
  • 协同编辑（多个客户端同时编辑文档，服务端同步变更）

Proto 定义

service ChatService {
  // 双向流式 RPC：聊天会话
  rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}

message ChatMessage {
  string user_id = 1;
  string room_id = 2;
  string content = 3;
  int64 timestamp = 4;
}

服务端实现（Go 示例）

func (s *chatServer) Chat(stream pb.ChatService_ChatServer) error {
    for {
        // 接收客户端消息
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        // 处理消息（例如：广播给房间内其他用户）
        reply := &pb.ChatMessage{
            UserId:    "system",
            RoomId:    msg.RoomId,
            Content:   fmt.Sprintf("用户 %s 说: %s", msg.UserId, msg.Content),
            Timestamp: time.Now().Unix(),
        }

        // 发送回复给客户端
        if err := stream.Send(reply); err != nil {
            return err
        }
    }
}

客户端调用（Go 示例）

stream, err := client.Chat(ctx)
if err != nil {
    log.Fatalf("创建聊天流失败: %v", err)
}

// 启动 goroutine 接收服务端消息
go func() {
    for {
        reply, err := stream.Recv()
        if err == io.EOF {
            return
        }
        if err != nil {
            log.Fatalf("接收失败: %v", err)
        }
        log.Printf("收到回复: %s", reply.Content)
    }
}()

// 主 goroutine 发送消息
messages := []string{"你好", "今天天气不错", "再见"}
for _, msg := range messages {
    chatMsg := &pb.ChatMessage{
        UserId:    "user-001",
        RoomId:    "room-001",
        Content:   msg,
        Timestamp: time.Now().Unix(),
    }
    if err := stream.Send(chatMsg); err != nil {
        log.Fatalf("发送失败: %v", err)
    }
    time.Sleep(1 * time.Second)
}

// 关闭发送端
stream.CloseSend()
// 等待接收 goroutine 完成
time.Sleep(2 * time.Second)

4.6 四种模式数据流对比图

下图直观展示了四种通信模式中客户端与服务端之间的数据流方向与交互时序：

gRPC 四种通信模式数据流对比 一元 RPC 客户端 → 服务端 → 客户端 客户端 服务端 请求 响应 服务端流式 RPC 客户端 → 服务端 → 客户端（多次） 客户端 服务端 请求 流式响应 客户端流式 RPC 客户端（多次）→ 服务端 → 客户端 客户端 服务端 流式请求 响应 双向流式 RPC 客户端 ↔ 服务端（独立双向） 客户端 服务端 双向流 说明：箭头方向表示数据流方向。一元 RPC 为单次请求-响应；服务端流式 RPC 为单次请求多次响应； 客户端流式 RPC 为多次请求单次响应；双向流式 RPC 为双方独立并发收发。

4.7 模式选择决策树

在实际微服务开发中，如何快速选择正确的通信模式？以下决策树可以帮助你做出判断：

gRPC 通信模式选择决策树 客户端需要流式传输？ 否 服务端需要流式响应？ 否 一元 RPC 是 服务端流式 RPC 是 服务端需要流式响应？ 否 客户端流式 RPC 是 双向流式 RPC 决策要点： 1. 客户端是否需要主动发送多条消息？是 → 考虑客户端流式或双向流式 2. 服务端是否需要主动推送多条消息？是 → 考虑服务端流式或双向流式 3. 双方都需要流式传输？是 → 双向流式 RPC

4.8 实践建议与避坑指南

性能考量：

  • 一元 RPC 适合延迟敏感、数据量小的场景，如用户认证、配置查询。
  • 流式 RPC 适合大数据量或实时推送场景，但会增加连接管理和背压处理的复杂度。
  • 双向流式 RPC 需要特别注意 goroutine 泄漏问题，确保在流关闭时正确清理资源。

常见陷阱：

  • 流未关闭： 客户端流式 RPC 必须调用 CloseAndRecv() 或 CloseSend()，否则服务端会一直等待。
  • 并发安全： 双向流式 RPC 的 Send() 和 Recv() 可以在不同 goroutine 中并发调用，但 Send() 本身不是线程安全的，需要加锁。
  • 超时设置： 流式 RPC 的超时时间应从流创建时开始计算，而不是每条消息。长时间运行的流应设置合理的超时或使用心跳机制。
  • 消息大小限制： gRPC 默认消息大小为 4MB，流式传输大消息时需调整 MaxRecvMsgSize 和 MaxSendMsgSize。

微服务架构建议： 在微服务内部通信中，优先使用一元 RPC 保持简单性；仅在确实需要流式传输的场景（如日志、监控、大文件）时才使用流式模式。过度使用流式 RPC 会增加系统复杂度和运维成本。

← 上一章 📖 返回目录 下一章 →