欢迎您访问新疆栾骏商贸有限公司，公司主营电子五金轴承产品批发业务！
全国咨询热线： 400-8878-609
新闻资讯
技术学院gRPC原生支持四种通信模式：Unary、Server Streaming、Client Streaming和Bidirectional Streaming。其中流式RPC适合大数据量、高实时性场景，能避免内存溢出、降低延迟、提升吞吐，并支持服务端推送与客户端持续发送。
gRPC 原生支持四种通信模式，其中流式 RPC（Streaming RPC）特别适合处理大数据量、实时性要求高或需要持续交互的场景。相比传统的一次请求-响应模型，流式传输能避免内存溢出、减少延迟、提升吞吐，并支持服务端推送、客户端持续发送等灵活交互方式。
gRPC 定义了以下四种流式通信方式，全部基于 HTTP/2 的多路复用和双向数据帧能力：
关键在于在 .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 { bo
ol 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() 等流控方法。
服务端需在一个 goroutine 中持续读取客户端消息，同时可随时写入响应。注意错误处理与连接生命周期管理：
ol accepted = 1;
string id = 2;
string reason = 3;
}
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 设备心跳或实时音视频元数据同步。