gRPC双向流模式允许客户端和服务端通过持久连接独立收发消息,适用于实时通信场景。在Go中,通过Protobuf定义stream类型接口,服务端使用Recv和Send处理流数据,客户端则用协程分离读写操作,需注意连接管理、错误重连与资源释放。

在分布式系统开发中,Go语言(Golang)凭借其轻量级的协程和高效的网络编程能力,成为构建高性能RPC服务的热门选择。而gRPC作为基于HTTP/2的远程过程调用框架,天然支持双向流通信,非常适合需要实时、持续数据交互的场景,比如聊天系统、实时推送、监控数据上报等。
理解gRPC中的双向流模式
gRPC定义了四种服务方法类型,其中双向流RPC是最灵活的一种:客户端和服务端都可以独立地发送和接收消息流,双方通过同一个持久连接进行全双工通信。
这种模式的关键在于:
- 调用开始后,客户端和服务端都可随时发送多个消息
- 消息顺序被保留,按发送顺序到达对端
- 连接由应用控制何时关闭,适合长期运行的数据通道
定义Protobuf接口实现双向流
要启用双向流,需在.proto文件中使用stream关键字声明请求和响应类型均为流式。
立即学习“go语言免费学习笔记(深入)”;
示例proto定义:
service ChatService {
rpc ChatStream(stream Message) returns (stream Message);
}
message Message {
string user = 1;
string content = 2;
int64 timestamp = 3;
}
上述定义表示ChatStream方法接收一个消息流,并返回一个消息流,是典型的双向通信结构。
服务端实现流处理逻辑
在Go中实现该接口时,gRPC会自动生成一个包含流对象的方法签名。服务端可通过读取和写入流来处理数据。
核心代码示例:
func (s *ChatServer) ChatStream(stream pb.ChatService_ChatStreamServer) error {
for {
// 接收来自客户端的消息
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
// 处理消息,例如广播给其他客户端
log.Printf("Received: %v", in.Content)
// 回复消息(可多次Send)
if err := stream.Send(&pb.Message{
User: "server",
Content: "echo: " + in.Content,
Timestamp: time.Now().Unix(),
}); err != nil {
return err
}
}
}
注意:Recv和Send可在同一协程交替调用,也可结合goroutine实现并发处理,但要注意并发安全。
客户端建立并使用双向流
客户端通过调用生成的Client方法获取流对象,之后即可同时发送和接收。
典型用法:
stream, err := client.ChatStream(context.Background())
if err != nil {
log.Fatal(err)
}
// 启动一个goroutine监听服务端消息
go func() {
for {
reply, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
log.Fatal(err)
}
log.Printf("Server: %s", reply.Content)
}
}()
// 主协程发送消息
for i := 0; i < 5; i++ {
msg := &pb.Message{
User: "client",
Content: fmt.Sprintf("msg %d", i),
Timestamp: time.Now().Unix(),
}
if err := stream.Send(msg); err != nil {
log.Fatal(err)
}
time.Sleep(time.Second)
}
stream.CloseSend()
这里使用两个协程分别处理发送与接收,避免阻塞。CloseSend通知服务端不再发送数据。
实际开发中的注意事项
双向流虽强大,但也需关注以下几点:
- 连接管理:长时间运行的流应加入心跳机制,防止空闲超时
- 错误处理:任一端出错或关闭都会导致整个流中断,需合理重连
- 资源释放:确保在defer中调用CloseSend或处理context取消
- 性能优化:大批量消息应考虑合并或限速,避免内存堆积










