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;
<p>service LogService {
// 双向流：客户端发送日志条目，服务端可实时反馈确认或过滤结果
rpc StreamLogs(stream LogEntry) returns (stream LogResponse);
}</p><p>message LogEntry {
string level = 1;
string message = 2;
int64 timestamp = 3;
}</p><p>message LogResponse {
bool accepted = 1;
string id = 2;
string reason = 3;
}
执行生成命令（需安装 protoc 和 protoc-gen-go、protoc-gen-go-grpc）：
立即学习“go语言免费学习笔记（深入）”；
protoc --go_out=. --go-grpc_out=. --go-grpc_opt=paths=source_relative logsvc.proto
生成的 Go 接口会包含 StreamLogs 方法，其参数为 LogService_StreamLogsServer（服务端）或 LogService_StreamLogsClient（客户端），均实现了 Recv()/Send() 等流控方法。
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等
服务端实现双向流逻辑（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)
}
<pre class="brush:php;toolbar:false;">// 处理单条日志（例如写入 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)
<p>stream, _ := client.StreamLogs(context.Background())
defer stream.CloseSend() // 发送端关闭，通知服务端“不再发了”</p><p>// 并发发送日志（可控制速率）
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) // 模拟节流
}
}()</p><p>// 同时接收服务端响应
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 设备心跳或实时音视频元数据同步。
