Golang如何使用gRPC Stream实现实时通信_Golang gRPC Stream数据流通信实践-Golang-PHP中文网

Golang如何使用gRPC Stream实现实时通信_Golang gRPC Stream数据流通信实践

P粉602998670

发布： 2025-12-02 18:35:38

原创

284人浏览过

gRPC支持四种通信模式，其中Server Streaming和Bi-directional Streaming适用于实时通信；通过定义.proto文件并生成Go代码，可实现服务端消息推送和双向聊天功能；服务端使用stream.Send发送消息，客户端循环Recv接收，结合context控制生命周期，注意错误处理、心跳机制与并发安全，即可构建高效实时通信系统。

golang如何使用grpc stream实现实时通信_golang grpc stream数据流通信实践

在构建高性能、低延迟的分布式系统时，实时通信是一个核心需求。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：

立即学习“go语言免费学习笔记（深入）”；

<font face="monospace">syntax = "proto3";
<p>package chat;</p><p>// 实时聊天服务
service ChatService {
// 客户端订阅消息流
rpc Subscribe(StreamRequest) returns (stream Message);
// 双向流聊天
rpc Chat(stream Message) returns (stream Message);
}</p><p>message StreamRequest {
string user_id = 1;
}</p><p>message Message {
string from = 1;
string content = 2;
int64 timestamp = 3;
}</font>

登录后复制

使用 protoc 生成 Go 代码：

<font face="monospace">protoc --go_out=. --go-grpc_out=. chat.proto</font>

登录后复制

会生成 chat.pb.go 和 chat_grpc.pb.go 文件。

实现 Server Streaming 实时推送

常见于服务端主动推送数据，如新闻广播、行情更新。

网易人工智能

网易数帆多媒体智能生产力平台

206

查看详情

服务端实现：

<font face="monospace">func (s *ChatServer) Subscribe(req *chat.StreamRequest, stream chat.ChatService_SubscribeServer) error {
  log.Printf("用户 %s 开始订阅", req.UserId)
<p>for i := 0; i < 10; i++ {
msg := &chat.Message{
From:      "system",
Content:   fmt.Sprintf("实时消息 #%d", i+1),
Timestamp: time.Now().Unix(),
}</p><pre class='brush:php;toolbar:false;'>// 发送消息到流
if err := stream.Send(msg); err != nil {
  return err
}
time.Sleep(1 * time.Second) // 模拟周期推送

登录后复制

} return nil }

客户端接收流：

<font face="monospace">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)
}</font>

登录后复制

这样客户端就能持续接收服务端推送的消息。

实现 Bi-directional Stream 实现聊天室

双向流允许客户端和服务端随时发送消息，适合即时通讯。

服务端处理双向流：

<font face="monospace">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
    }
<pre class='brush:php;toolbar:false;'>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
}

登录后复制

} }

客户端也可以一边发一边收：

<font face="monospace">stream, _ := client.Chat(context.Background())
<p>// 启动 goroutine 接收消息
go func() {
for {
msg, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
log.Fatal(err)
}
log.Printf("回复: %s", msg.Content)
}
}()</p><p>// 发送消息
for i := 1; i <= 5; i++ {
msg := &chat.Message{
From:    "client",
Content: fmt.Sprintf("第 %d 条消息", i),
}
stream.Send(msg)
time.Sleep(2 * time.Second)
}</font>

登录后复制

这种模式下，通信是异步且持续的。