gRPC原生支持四种通信模式：Unary、Server Streaming、Client Streaming和Bidirectional Streaming。其中流式RPC适合大数据量、高实时性场景，能避免内存溢出、降低延迟、提升吞吐，并支持服务端推送与客户端持续发送。
gRPC 原生支持四种通信模式，其中流式 RPC（Streaming RPC）特别适合处理大数据量、实时性要求高或需要持续交互的场景。相比传统的一次请求-响应模型，流式传输能避免内存溢出、减少延迟、提升吞吐，并支持服务端推送、客户端持续发送等灵活交互方式。
理解 gRPC 的四种流类型
gRPC 定义了以下四种流式通信方式，全部基于 HTTP/2 的多路复用和双向数据帧能力：
- Unary（一元）：最常见，客户端发一次，服务端回一次（非流）
- Server Streaming（服务端流）：客户端发一次请求，服务端返回多个响应（如日志尾部、实时指标推送）
- Client Streaming（客户端流）：客户端连续发送多个请求，服务端汇总后统一响应（如上传大文件分块、语音流识别）
- Bidirectional Streaming（双向流）：双方均可随时收发消息，完全异步（如聊天室、实时协同编辑）
定义 .proto 文件并生成 Go 代码
关键在于在 .proto 文件中使用 stream 关键字声明流式方法。例如实现一个双向流式日志转发服务：
syntax = "proto3"; package logsvc;service LogService { // 双向流：客户端发送日志条目，服务端可实时反馈确认或过滤结果 rpc StreamLogs(stream LogEntry) returns (stream LogResponse); }
message LogEntry { string level = 1; string message = 2; int64 timestamp = 3; }
message LogResponse { bool accepted = 1; string id = 2; string reason = 3; }
执行生成命令（需安装 protoc 和 protoc-gen-go、protoc-gen-go-grpc）：
protoc --go_out=. --go-grpc_out=. --go-grpc_opt=paths=source_relative logsvc.proto
生成的 Go 接口会包含 StreamLogs 方法，其参数为 LogService_StreamLogsServer（服务端）或 LogService_StreamLogsClient（客户端），均实现了 Recv()/Send() 等流控方法。
服务端实现双向流逻辑（Go）
服务端需在一个 goroutine 中持续读取客户端消息，同时可随时写入响应。注意错误处理与连接生命周期管理：
func (s *logServer) StreamLogs(stream logsvc.LogService_StreamLogsServer) error{ for { req, err := stream.Recv() if err == io.EOF { return nil // 客户端关闭流 } if err != nil { return status.Errorf(codes.Unknown, "recv failed: %v", err) }
// 处理单条日志（例如写入 Kafka、校验格式、异步落盘） resp := &logsvc.LogResponse{ Accepted: true, Id: fmt.Sprintf("log-%d", time.Now().UnixNano()), } // 异步响应（不阻塞接收）——可配合 select + channel 控制背压 if err := stream.Send(resp); err != nil { return status.Errorf(codes.Unavailable, "send failed: %v", err) }} }
⚠️ 注意：
Recv()是阻塞调用；若需并发处理（如批量聚合后再响应），建议将接收的消息发到内部 channel，由 worker goroutine 消费。客户端发起流式调用（Go）
客户端同样使用
Send()和Recv()，但顺序和节奏由业务决定。例如模拟持续发送日志：conn, _ := grpc.Dial("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials())) defer conn.Close() client := logsvc.NewLogServiceClient(conn)stream, _ := client.StreamLogs(context.Background()) defer stream.CloseSend() // 发送端关闭，通知服务端“不再发了”
// 并发发送日志（可控制速率） go func() { for i := 0; i < 100; i++ { entry := &logsvc.LogEntry{ Level: "INFO", Message: fmt.Sprintf("log #%d", i), Timestamp: time.Now().Unix(), } if err := stream.Send(entry); err != nil { log.Printf("send error: %v", err) return } time.Sleep(10 * time.Millisecond) // 模拟节流 } }()
// 同时接收服务端响应 for { resp, err := stream.Recv() if err == io.EOF { break } if err != nil { log.Printf("recv error: %v", err) break } log.Printf("Got response: %+v", resp) }
? 小技巧：用
context.WithTimeout或WithCancel控制整个流生命周期；对超大数据流，可结合runtime.GC()或debug.FreeOSMemory()（谨慎使用）缓解内存压力。基本上就这些。流式 RPC 不是魔法，核心在于理解流的边界（何时 EOF）、错误传播机制（单次 Send/Recv 失败是否终止整个流）、以及如何与业务逻辑解耦（比如用 channel 缓冲、用 worker 池处理）。只要协议定义清晰、流控得当，gRPC 流完全能扛住 GB 级日志、百万级 IoT 设备心跳或实时音视频元数据同步。
{
for {
req, err := stream.Recv()
if err == io.EOF {
return nil // 客户端关闭流
}
if err != nil {
return status.Errorf(codes.Unknown, "recv failed: %v", err)
}
