gRPC支持四种通信模式，其中Server Streaming和Bi-directional Streaming适用于实时通信；通过定义.proto文件并生成Go代码，可实现服务端消息推送和双向聊天功能；服务端使用stream.Send发送消息，客户端循环Recv接收，结合context控制生命周期，注意错误处理、心跳机制与并发安全，即可构建高效实时通信系统。
在构建高性能、低延迟的分布式系统时，实时通信是一个核心需求。gRPC 作为 Google 推出的高性能 RPC 框架，原生支持 Stream（流）模式，非常适合用于实现实时数据传输。Golang 结合 gRPC Stream 可以轻松实现客户端与服务端之间的双向持续通信，比如聊天系统、实时通知、监控推送等场景。
理解 gRPC 的四种通信模式
gRPC 支持四种调用方式，其中三种涉及流（Stream），是实现实时通信的基础：
- Unary RPC：传统的一次请求一次响应。
- Server Streaming RPC：客户端发送一次请求，服务端返回一个流，持续发送多个消息。
- Client Streaming RPC：客户端通过流发送多个消息，服务端最终返回一次响应。
- Bi-directional Streaming RPC：双方都使用流，可同时收发消息，适合全双工通信。
实时通信通常选择 Server Stream 或 Bi-directional Stream。
定义 .proto 文件并生成代码
以一个简单的实时消息推送为例，定义 chat.proto：
syntax = "proto3";package chat;
// 实时聊天服务 service ChatService { // 客户端订阅消息流 rpc Subscribe(StreamRequest) returns (stream Message); // 双向流聊天 rpc Chat(stream Message) returns (stream Message); }
message StreamRequest { string user_id = 1; }
message Message { string from = 1; string content = 2; int64 timestamp = 3; }
使用 protoc 生成 Go 代码：
protoc --go_out=. --go-grpc_out=. chat.proto
会生成 chat.pb.go 和 chat_grpc.pb.go 文件。
实现 Server Streaming 实时推送
常见于服务端主动推送数据，如新闻广播、行情更新。
服务端实现：
func (s *ChatServer) Subscribe(req *chat.StreamRequest, stream chat.ChatService_SubscribeServer) error {
log.Printf("用户 %s 开始订阅", req.UserId)
for i := 0; i < 10; i++ {
msg := &chat.Message{
From: "system",
Content: fmt.Sprintf("实时消息 #%d", i+1),
Timestamp: time.Now().Unix(),
}
// 发送消息到流
if err := stream.Send(msg); err != nil {
return err
}
time.Sleep(1 * time.Second) // 模拟周期推送}
return nil
}
客户端接收流：
stream, _ := client.Subscribe(context.Background(), &chat.StreamRequest{UserId: "user123"})
for {
msg, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
log.Fatal(err)
}
log.Printf("收到消息: %s (来自 %s)", msg.Content, msg.From)
}这样客户端就能持续接收服务端推送的消息。
实现 Bi-directional Stream 实现聊天室
双向流允许客户端和服务端随时发送消息，适合即时通讯。
服务端处理双向流：
func (s *ChatServer) Chat(stream chat.ChatService_ChatServer) error {
for {
// 接收客户端消息
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
log.Printf("收到消息
: %s (来自 %s)", in.Content, in.From)
// 回复消息
out := &chat.Message{
From: "server",
Content: "已收到: " + in.Content,
Timestamp: time.Now().Unix(),
}
if err := stream.Send(out); err != nil {
return err
}}
}
客户端也可以一边发一边收：
stream, _ := client.Chat(context.Background())
// 启动 goroutine 接收消息
go func() {
for {
msg, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
log.Fatal(err)
}
log.Printf("回复: %s", msg.Content)
}
}()
// 发送消息
for i := 1; i <= 5; i++ {
msg := &chat.Message{
From: "client",
Content: fmt.Sprintf("第 %d 条消息", i),
}
stream.Send(msg)
time.Sleep(2 * time.Second)
}
这种模式下，通信是异步且持续的。
注意事项与最佳实践
-
错误处理：流过程中网络中断或超时常见，需做好重连机制。
-
上下文控制：使用 context 控制流的生命周期，避免 goroutine 泄漏。
-
心跳机制：长时间空闲可能被中间代理断开，建议定期发送心跳消息。
-
并发安全：多个 goroutine 操作同一 stream 时需注意同步问题。
-
流状态管理：服务端可维护连接列表，实现广播或多播逻辑。
基本上就这些。Golang + gRPC Stream 提供了一套简洁高效的实时通信方案，合理使用能极大提升系统响应能力。关键是理解流的生命周期和控制方式，结合业务设计好通信协议。
: %s (来自 %s)", in.Content, in.From)
// 回复消息
out := &chat.Message{
From: "server",
Content: "已收到: " + in.Content,
Timestamp: time.Now().Unix(),
}
if err := stream.Send(out); err != nil {
return err
}
