gRPC的双向流通信模式允许客户端和服务端建立连接后，双方都可以独立地向对方发送消息流，不需要等待对方的响应，非常适合实时数据同步、长连接推送等场景。在Golang中实现gRPC双向流，需要先定义protobuf协议，再分别实现服务端和客户端的逻辑。
环境准备
首先需要安装Golang环境和相关依赖工具：
- 安装Golang，版本建议1.16及以上
- 安装protobuf编译器protoc，用于生成Golang代码
- 安装Golang的protobuf和gRPC相关插件：
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest go get google.golang.org/grpc go get google.golang.org/protobuf
定义protobuf协议文件
双向流通信需要在protobuf的rpc定义中使用stream关键字标记请求和响应参数，下面定义一个简单的聊天场景的协议文件chat.proto：
syntax = "proto3";
package chat;
option go_package = "./chat";
// 定义聊天消息结构
message ChatMessage {
string sender = 1; // 发送者
string content = 2; // 消息内容
int64 timestamp = 3; // 时间戳
}
// 定义聊天服务
service ChatService {
// 双向流通信的rpc方法，请求和响应都带stream
rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}
执行以下命令生成Golang代码：
protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative chat.proto
实现gRPC服务端
服务端需要实现ChatServiceServer接口，处理客户端的双向流请求，这里实现简单的消息转发逻辑，将客户端发送的消息加上服务端标识后返回：
package main
import (
"context"
"fmt"
"io"
"log"
"net"
"time"
pb "your_module_path/chat" // 替换为实际的模块路径
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// 定义服务端结构体，实现ChatServiceServer接口
type chatServer struct {
pb.UnimplementedChatServiceServer
}
// 实现Chat方法，处理双向流
func (s *chatServer) Chat(stream pb.ChatService_ChatServer) error {
// 启动一个goroutine接收客户端发送的消息
go func() {
for {
// 接收客户端消息
in, err := stream.Recv()
if err == io.EOF {
// 客户端关闭发送流
return
}
if err != nil {
log.Printf("接收客户端消息失败: %v", err)
return
}
log.Printf("收到客户端消息: 发送者=%s, 内容=%s", in.Sender, in.Content)
// 构造响应消息，加上服务端标识
resp := &pb.ChatMessage{
Sender: "server",
Content: fmt.Sprintf("已收到你的消息: %s", in.Content),
Timestamp: time.Now().Unix(),
}
// 发送响应给客户端
if err := stream.Send(resp); err != nil {
log.Printf("发送消息给客户端失败: %v", err)
return
}
}
}()
// 服务端也可以主动发送消息给客户端，这里每隔3秒发送一条心跳消息
ticker := time.NewTicker(3 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
heartbeat := &pb.ChatMessage{
Sender: "server",
Content: "服务端心跳消息",
Timestamp: time.Now().Unix(),
}
if err := stream.Send(heartbeat); err != nil {
log.Printf("发送心跳消息失败: %v", err)
return status.Error(codes.Aborted, "连接中断")
}
case <-stream.Context().Done():
// 连接上下文结束，退出循环
return nil
}
}
}
func main() {
// 监听端口
lis, err := net.Listen("tcp", ":50051")
if err != nil {
log.Fatalf("监听端口失败: %v", err)
}
// 创建gRPC服务端
s := grpc.NewServer()
// 注册服务
pb.RegisterChatServiceServer(s, &chatServer{})
log.Printf("gRPC服务端启动，监听端口: 50051")
// 启动服务
if err := s.Serve(lis); err != nil {
log.Fatalf("启动服务失败: %v", err)
}
}
实现gRPC客户端
客户端需要调用Chat方法获取双向流对象，然后分别启动goroutine处理发送和接收消息的逻辑：
package main
import (
"context"
"io"
"log"
"time"
pb "your_module_path/chat" // 替换为实际的模块路径
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
)
func main() {
// 连接服务端
conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
log.Fatalf("连接服务端失败: %v", err)
}
defer conn.Close()
// 创建客户端
client := pb.NewChatServiceClient(conn)
// 调用Chat方法获取双向流
stream, err := client.Chat(context.Background())
if err != nil {
log.Fatalf("调用Chat方法失败: %v", err)
}
// 启动goroutine接收服务端消息
go func() {
for {
resp, err := stream.Recv()
if err == io.EOF {
log.Println("服务端关闭发送流")
return
}
if err != nil {
log.Printf("接收服务端消息失败: %v", err)
return
}
log.Printf("收到服务端消息: 发送者=%s, 内容=%s", resp.Sender, resp.Content)
}
}()
// 客户端发送消息
messages := []string{"你好，服务端", "这是第二条消息", "测试双向流通信"}
for _, msg := range messages {
sendMsg := &pb.ChatMessage{
Sender: "client",
Content: msg,
Timestamp: time.Now().Unix(),
}
if err := stream.Send(sendMsg); err != nil {
log.Fatalf("发送消息失败: %v", err)
}
log.Printf("客户端发送消息: %s", msg)
time.Sleep(2 * time.Second)
}
// 关闭客户端的发送流
if err := stream.CloseSend(); err != nil {
log.Printf("关闭发送流失败: %v", err)
}
// 等待接收完所有消息
time.Sleep(5 * time.Second)
}
运行测试
先启动服务端程序，再启动客户端程序，可以看到双方都能独立发送和接收消息，服务端会定时发送心跳，客户端发送的消息也会被服务端处理后返回。需要注意的是，双向流的关闭需要双方都处理对应的EOF信号，避免连接泄漏。
注意事项
- 双向流的生命周期和gRPC连接绑定，连接断开后流也会自动关闭
- 发送和接收操作需要做好并发控制，避免同时操作流导致的错误
- 实际项目中需要添加更完善的错误处理和重连机制
- 流中的消息顺序是有保障的，同一方向发送的消息会按发送顺序被接收