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

Golang双向流式RPC实现详解

2025-09-05 08:08:31 0浏览收藏

大家好，今天本人给大家带来文章《Golang实现双向流式RPC详解》，文中内容主要涉及到，如果你对Golang方面的知识点感兴趣，那就请各位朋友继续看下去吧~希望能真正帮到你们，谢谢！

双向流式RPC在Golang中通过客户端和服务端持续发送多个消息实现灵活通信。定义proto接口时，使用stream关键字声明参数和返回值，如rpc BidirectionalChat(stream MessageRequest) returns (stream MessageResponse)。服务端逻辑需处理Recv()接收客户端消息并Send()回应。客户端则创建流对象后用goroutine监听服务端响应，主线程发送消息。注意上下文控制、错误处理及测试调试以确保稳定交互。

如何用Golang实现双向流式RPC 深入gRPC流式通信模式

在使用 Golang 实现 gRPC 的时候，很多人会接触到“流式 RPC”这个概念。其中双向流式 RPC 是最灵活、也最具挑战的一种通信模式。它允许客户端和服务端各自持续发送多个消息，形成一个双向流动的通道。这种机制非常适合实时性要求高、交互频繁的场景，比如聊天应用、实时数据推送等。

要实现这样的通信方式，关键在于定义好 proto 接口，并在服务端和客户端正确处理流的读写操作。

定义 proto 文件：双向流式接口的基础

gRPC 的四种调用方式中，双向流式是通过 rpc 方法中参数和返回值都使用 stream 关键字来声明的。例如：

syntax = "proto3";

package chat;

service ChatService {
  rpc BidirectionalChat(stream MessageRequest) returns (stream MessageResponse);
}

message MessageRequest {
  string content = 1;
}

message MessageResponse {
  string reply = 1;
}

上面的例子中，BidirectionalChat 就是一个典型的双向流方法。客户端可以持续发送 MessageRequest，服务端也能持续返回 MessageResponse。proto 文件定义清楚后，就可以用 protoc 工具生成对应的服务代码。

注意：确保你安装了 protoc-gen-go-grpc 插件，否则无法生成正确的流式接口代码。

编写服务端逻辑：接收请求并响应流

在 Golang 中实现服务端时，gRPC 自动生成的代码会提供一个带有两个流参数的方法签名。你需要在这个方法内部处理输入流和输出流。

举个例子：

func (s *chatServer) BidirectionalChat(stream pb.ChatService_BidirectionalChatServer) error {
    for {
        req, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        // 处理客户端发来的消息
        log.Printf("收到消息: %s", req.Content)

        // 回复客户端
        resp := &pb.MessageResponse{
            Reply: fmt.Sprintf("服务器收到了: %s", req.Content),
        }
        if err := stream.Send(resp); err != nil {
            return err
        }
    }
}

这段代码的核心是不断从客户端读取消息（Recv），然后根据内容做出回应（Send）。需要注意的是，一旦客户端关闭连接（如遇到 EOF），服务端也应该优雅退出该函数。

客户端实现：双向发送与接收消息

客户端方面，使用 gRPC 生成的 BidirectionalChatClient 类型创建一个流对象，之后就可以同时发送和接收消息了。

示例代码如下：

stream, err := client.BidirectionalChat(context.Background())
if err != nil {
    log.Fatalf("无法建立流: %v", err)
}

// 单独开一个 goroutine 来监听服务端消息
go func() {
    for {
        resp, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatalf("接收失败: %v", err)
        }
        fmt.Println("收到回复:", resp.Reply)
    }
}()

// 主线程循环发送消息
for i := 0; i < 5; i++ {
    msg := fmt.Sprintf("第 %d 条消息", i+1)
    if err := stream.Send(&pb.MessageRequest{Content: msg}); err != nil {
        log.Fatalf("发送失败: %v", err)
    }
    time.Sleep(time.Second)
}

// 发送完后关闭流
stream.CloseSend()

这里用了 goroutine 来并发处理接收服务端的消息，而主线程则负责定时发送消息。这种方式适用于需要长时间保持连接并双向通信的场景。

此外，要注意上下文（context）控制和错误处理，避免因为网络问题或对方断开导致程序卡住。