4. gRPC 服务端流代码实战（Go）：定义 proto 文件、生成 Go 代码、实现流式服务端、编写客户端接收流

本章将带领你从零开始，使用 Go 语言实现一个完整的 gRPC 服务端流模式。我们将依次完成 Proto 文件定义、Go 代码生成、流式服务端实现 以及 客户端流接收 四个核心步骤。通过本实战，你将掌握服务端流的核心开发流程与关键细节。

4.1 定义 Proto 文件：声明服务端流 RPC

服务端流 RPC 的关键在于 响应消息前使用 stream 关键字。以下示例定义了一个 StockService，客户端发送一个股票代码请求，服务端持续返回实时价格流。

syntax = "proto3";

package stock;

option go_package = "github.com/your-project/stockpb";

// 请求：单个股票代码
message StockRequest {
  string symbol = 1;
}

// 响应：实时价格数据
message StockPrice {
  string symbol = 1;
  double price = 2;
  int64 timestamp = 3;
}

// 服务定义：服务端流 RPC
service StockService {
  // 客户端发送一个请求，服务端返回价格流
  rpc SubscribeStock(StockRequest) returns (stream StockPrice);
}

4.2 生成 Go 代码：protoc 编译与产物结构

在项目根目录执行以下命令，生成 Go 代码：

protoc --go_out=. --go-grpc_out=. \
  --go_opt=paths=source_relative \
  --go-grpc_opt=paths=source_relative \
  proto/stock.proto

执行后，proto/ 目录下会生成两个文件：

• stock.pb.go：消息类型的 Go 结构体（StockRequest、StockPrice）及序列化方法。
• stock_grpc.pb.go：服务端接口、客户端桩代码及注册函数。

SubscribeStock(*StockRequest, StockService_SubscribeStockServer) error

4.3 实现流式服务端：发送数据流

服务端需要实现 SubscribeStock 方法，在循环中通过 stream.Send() 持续推送数据。以下是一个模拟实时股票价格的实现：

type stockServer struct {
  stockpb.UnimplementedStockServiceServer
}

func (s *stockServer) SubscribeStock(
  req *stockpb.StockRequest,
  stream stockpb.StockService_SubscribeStockServer,
) error {
  symbol := req.GetSymbol()
  log.Printf("订阅股票: %s", symbol)

  // 模拟持续发送价格数据
  for i := 0; i < 10; i++ {
    price := &stockpb.StockPrice{
      Symbol:    symbol,
      Price:     100.0 + float64(i)*0.5,
      Timestamp: time.Now().UnixMilli(),
    }
    // 发送消息到流
    if err := stream.Send(price); err != nil {
      log.Printf("发送失败: %v", err)
      return err
    }
    time.Sleep(1 * time.Second) // 模拟实时间隔
  }
  return nil
}

4.4 编写客户端接收流：逐条处理

客户端通过 Recv() 方法循环读取服务端推送的每条消息，直到流结束或发生错误。

func main() {
  conn, _ := grpc.Dial("localhost:50051", grpc.WithInsecure())
  defer conn.Close()

  client := stockpb.NewStockServiceClient(conn)
  req := &stockpb.StockRequest{Symbol: "GOOG"}

  // 调用服务端流 RPC，返回一个流对象
  stream, err := client.SubscribeStock(context.Background(), req)
  if err != nil {
    log.Fatalf("调用失败: %v", err)
  }

  // 循环接收流消息
  for {
    price, err := stream.Recv()
    if err == io.EOF {
      log.Println("流结束")
      break
    }
    if err != nil {
      log.Fatalf("接收错误: %v", err)
    }
    log.Printf("收到价格: %s = %.2f @ %d",
      price.GetSymbol(), price.GetPrice(), price.GetTimestamp())
  }
}

4.5 服务端流通信流程总览

下图展示了服务端流 RPC 的完整通信流程，从客户端发起请求到流结束：

4.6 代码结构对比：一元 RPC vs 服务端流 RPC

4.7 常见错误与避坑指南

• 忘记处理 io.EOF：客户端 Recv() 返回 io.EOF 是正常结束信号，不要当作错误处理。
• 服务端未关闭流：方法返回后，gRPC 框架会自动关闭流。不要在 Send() 后手动调用 CloseSend()（那是客户端用的）。
• 阻塞问题：如果服务端发送速度远快于客户端处理速度，Send() 会阻塞。建议在服务端加入速率限制或背压感知逻辑。
• Context 取消：客户端取消请求时，服务端应通过 stream.Context().Done() 检测并退出循环，避免资源泄漏。

select {
case <-stream.Context().Done():
    return stream.Context().Err()
default:
    // 继续发送
}