当前位置：首页 > 文章列表 > Golang > Go教程 > GolangRPC双向流通信实现详解

GolangRPC双向流通信实现详解

2025-12-14 23:58:42 0浏览收藏

珍惜时间，勤奋学习！今天给大家带来《Golang RPC双向流通信实现教程》，正文内容主要涉及到等等，如果你正在学习Golang，或者是对Golang有疑问，欢迎大家关注我！后面我会持续更新相关内容的，希望都能帮到正在学习的大家！

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

GolangRPC双向流通信开发实践

在分布式系统开发中，Go语言（Golang）凭借其轻量级的协程和高效的网络编程能力，成为构建高性能RPC服务的热门选择。而gRPC作为基于HTTP/2的远程过程调用框架，天然支持双向流通信，非常适合需要实时、持续数据交互的场景，比如聊天系统、实时推送、监控数据上报等。

理解gRPC中的双向流模式

gRPC定义了四种服务方法类型，其中双向流RPC是最灵活的一种：客户端和服务端都可以独立地发送和接收消息流，双方通过同一个持久连接进行全双工通信。

这种模式的关键在于：

调用开始后，客户端和服务端都可随时发送多个消息
消息顺序被保留，按发送顺序到达对端
连接由应用控制何时关闭，适合长期运行的数据通道

定义Protobuf接口实现双向流

要启用双向流，需在.proto文件中使用stream关键字声明请求和响应类型均为流式。

示例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通知服务端不再发送数据。