522 lines
17 KiB
Markdown
522 lines
17 KiB
Markdown
---
|
||
tags: [gRPC, Go, Streaming, RPC]
|
||
create time: 2026-05-11 16:30
|
||
---
|
||
|
||
# Streaming Handler
|
||
|
||
## 概述
|
||
|
||
本文档讲解 gRPC 三种 streaming RPC handler 的编写模式——**Server Streaming**、**Client Streaming** 和 **Bidirectional Streaming**。核心主题是循环内的生命周期管理：如何正确感知 client 断开、如何在大数据量下避免 goroutine 泄漏、以及如何优雅地处理背压。
|
||
|
||
> [!warning] Streaming handler 最大的陷阱：goroutine 泄漏
|
||
> 如果你在 loop 里 spawn 了子 goroutine 但没有正确的退出机制，一旦 client 断开连接，server 端的 goroutine 会永远无法回收。**下面每一个例子都会标注如何避免这个问题。**
|
||
|
||
## Server Streaming Handler
|
||
|
||
Server streaming：client 发一次请求，server 持续返回多条响应。
|
||
|
||
```go
|
||
func (s *server) ListUsers(req *pb.ListUsersRequest, stream pb.UserService_ListUsersServer) error {
|
||
users := getAllUsers() // 从 DB / cache 获取
|
||
|
||
for _, u := range users {
|
||
if err := stream.Send(&pb.ListUsersResponse{User: u}); err != nil {
|
||
// client 已经断开或主动取消
|
||
s.logger.Warn("send failed", "error", err)
|
||
return status.Error(codes.Internal, err.Error())
|
||
}
|
||
}
|
||
return nil // 正常结束：数据全部发送完毕
|
||
}
|
||
```
|
||
|
||
### 核心要点
|
||
|
||
- Server 通过 `stream.Send()` 逐条发送，每次 Send 都会编码 Protobuf 并通过 HTTP/2 frame 推送到 client
|
||
- **`stream.Context()` 是唯一可靠的退出信号** —— 当 client cancel 或网络中断时，context 会被取消
|
||
- 正常结束（遍历完全部数据）返回 `nil`；断连时 context 报错
|
||
- 不要在 loop 之外做 cleanup，否则即使 send 失败也会执行不必要的清理
|
||
|
||
### 大数据量场景下的超时检测
|
||
|
||
上面的简单示例只适用于结果集可控的场景。如果用户量大（比如百万级），需要在循环中**主动检查 context** 提前退出：
|
||
|
||
```go
|
||
func (s *server) ListUsers(req *pb.ListUsersRequest, stream pb.UserService_ListUsersServer) error {
|
||
ctx := stream.Context()
|
||
results := hugeQuery(ctx) // 可能返回大量数据
|
||
|
||
for i, u := range results {
|
||
// 每发送 100 条检查一下 context
|
||
if i >= 100 && len(results) >= 1000 {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err() // client 已取消
|
||
default:
|
||
}
|
||
}
|
||
|
||
if err := stream.Send(&pb.ListUsersResponse{User: u}); err != nil {
|
||
return status.Error(codes.Internal, err.Error())
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
```
|
||
|
||
这里有个设计取舍：**每次 Send 都检查 vs 每隔 N 次检查**。全检查保证响应快但多了 select 开销；间隔检查减少 CPU 消耗但有延迟。根据业务对"取消感知的灵敏度"来决定。
|
||
|
||
### 数据流转时序
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant C as Client
|
||
participant S as Server Handler
|
||
participant DB as Data Source
|
||
|
||
C->>S: Send(ListUsersRequest)
|
||
S->>DB: Query All Users
|
||
DB-->>S: []User
|
||
|
||
loop For Each User
|
||
alt Client Still Connected
|
||
S->>S: Check Context
|
||
S->>C: Send(User)
|
||
else Context Cancelled
|
||
S->>S: Return ctx.Err()
|
||
end
|
||
end
|
||
|
||
alt Normal Completion
|
||
C->>S: (no more data)
|
||
S-->>C: Return nil (EOF)
|
||
else Send Failed
|
||
S-->>C: Return error Status
|
||
end
|
||
```
|
||
|
||
## Client Streaming Handler
|
||
|
||
Client streaming：client 持续发送多条请求，server 汇总后返回一次响应。
|
||
|
||
```go
|
||
func (s *server) UploadData(stream pb.UserService_UploadDataServer) (*pb.UploadResult, error) {
|
||
ctx := stream.Context()
|
||
var totalBytes int64
|
||
var processedItems []*Item
|
||
|
||
for {
|
||
chunk, err := stream.Recv()
|
||
if err == io.EOF {
|
||
break // client 结束了发送
|
||
}
|
||
if err != nil {
|
||
// client 主动取消或网络异常 —— 不算业务错误，直接退出
|
||
if ctx.Err() != nil {
|
||
return nil, ctx.Err()
|
||
}
|
||
return nil, status.Errorf(codes.InvalidArgument, "recv failed: %v", err)
|
||
}
|
||
|
||
totalBytes += int64(len(chunk.Data))
|
||
processedItems = append(processedItems, chunk.ToItem())
|
||
// TODO: 批量写入 DB，不要每来一条就 flush
|
||
}
|
||
|
||
return &pb.UploadResult{TotalBytes: totalBytes, Count: int32(len(processedItems))}, nil
|
||
}
|
||
```
|
||
|
||
### 核心要点
|
||
|
||
- 用 `io.EOF` 判断 client 结束发送（注意：是 `io.EOF`，不是其他 error）
|
||
- **Recv 的 error 可能来自两种情况**：client 正常关闭（`io.EOF`）或 client 取消/断连。后者不应视为业务错误，而应直接通过 `ctx.Err()` 退出
|
||
- Recv 循环必须在 EOF 之前 `return` 或 `break`，否则会无限阻塞
|
||
- 所有临时状态放在 loop 内，返回前一次性聚合结果
|
||
|
||
### 内存安全注意事项
|
||
|
||
当 client 发送的数据量不可控时，**必须限制内存使用**：
|
||
|
||
```go
|
||
const maxBufferSize = 100 * 1024 * 1024 // 100 MB limit
|
||
|
||
func (s *server) UploadData(stream pb.UserService_UploadDataServer) (*pb.UploadResult, error) {
|
||
var totalBytes int64
|
||
var items []*Item // ⚠️ 生产环境应写流到磁盘而非放内存
|
||
|
||
for {
|
||
chunk, err := stream.Recv()
|
||
if err == io.EOF {
|
||
break
|
||
}
|
||
if err != nil {
|
||
return nil, status.Errorf(codes.InvalidArgument, "recv failed: %v", err)
|
||
}
|
||
|
||
totalBytes += int64(len(chunk.Data))
|
||
if totalBytes > maxBufferSize {
|
||
return nil, status.Errorf(codes.ResourceExhausted,
|
||
"exceeds max upload size: %d bytes", maxBufferSize)
|
||
}
|
||
|
||
items = append(items, chunk.ToItem())
|
||
}
|
||
|
||
return aggregateResult(items), nil
|
||
}
|
||
```
|
||
|
||
> [!tip] 何时该用临时文件？
|
||
> 如果单条消息超过 10MB 或者总量难以预估，直接将 `chunk.Data` 写到一个 `os.File` 中比存在内存切片里安全得多。handler 返回前一次性处理临时文件即可。
|
||
|
||
## Bidirectional Streaming Handler
|
||
|
||
Bidirectional streaming 是最复杂的模式：收和发两个方向完全独立，各自有自己的生命周期。
|
||
|
||
```go
|
||
func (s *server) ChatRoom(stream pb.UserService_ChatRoomServer) error {
|
||
ctx := stream.Context()
|
||
inboundCh := make(chan *pb.ChatMessage, 100)
|
||
errCh := make(chan error, 1) // 错误传播通道
|
||
|
||
// ---- 接收方向（后台 goroutine）----
|
||
go func() {
|
||
defer close(inboundCh)
|
||
for {
|
||
msg, err := stream.Recv()
|
||
if err != nil {
|
||
errCh <- err // EOF / cancel / network error
|
||
return
|
||
}
|
||
select {
|
||
case inboundCh <- msg:
|
||
default:
|
||
// channel full, drop message
|
||
}
|
||
}
|
||
}()
|
||
|
||
// ---- 发送方向（主循环）----
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err() // client 断了
|
||
case err := <-errCh:
|
||
return err // recv 出错，主循环退出
|
||
case envelope := <-inboundCh:
|
||
envelope := transform(envelope) // 业务转换
|
||
if err := stream.Send(envelope); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
### 核心架构
|
||
|
||
```mermaid
|
||
flowchart TB
|
||
subgraph "Handler Goroutine"
|
||
S["Send Loop<br/>for-select"]
|
||
end
|
||
|
||
subgraph "Recv Goroutine"
|
||
R["Recv Loop"] -->|"inboundCh buffered"| B["Business Logic"]
|
||
end
|
||
|
||
R --> stream
|
||
S --> stream
|
||
B -->|"outboundCh buffered"| S
|
||
|
||
stream["HTTP/2 Stream"]
|
||
|
||
style S fill:#4FC08D,color:#fff
|
||
style R fill:#00B6BC,color:#fff
|
||
style B fill:#FF9F43,color:#000
|
||
```
|
||
|
||
### 关键设计原则
|
||
|
||
1. **Context.Done() 是唯一可靠的退出信号** — 不管哪个方向出错，最终都要通过 context 来协调
|
||
2. **收和发是两个独立的生命周期** — recv 在一个 goroutine，send 在主循环
|
||
3. **务必给 channel 设置 buffer**，否则 sender 堵住时 receiver 也会饿死
|
||
|
||
### 背压策略选择
|
||
|
||
channel full 时如何处理，决定了系统的韧性：
|
||
|
||
| 策略 | 实现方式 | 适用场景 | 风险 |
|
||
|------|---------|---------|------|
|
||
| **Drop** | `select + default`（如上文） | 实时性优先，丢消息可接受 | 丢失关键消息 |
|
||
| **Block** | 普通 `ch <- msg` | 吞吐量重要，允许等待 | receiver 被拖慢 |
|
||
| **Backpressure** | `select + ctx.Done()` | 需要优雅降级 | 代码略复杂 |
|
||
| **Reject** | 统计丢弃数，超标后返回 error | 有 SLA 要求的系统 | 直接中断连接 |
|
||
|
||
**推荐**大多数场景用 Backpressure 模式——超时未消费则丢弃，连续丢弃过多时通过 errCh 通知主循环终止：
|
||
|
||
```go
|
||
func (s *server) ChatRoom(stream pb.UserService_ChatRoomServer) error {
|
||
ctx := stream.Context()
|
||
inboundCh := make(chan *pb.ChatMessage, 100)
|
||
errCh := make(chan error, 1)
|
||
|
||
go func() {
|
||
defer close(inboundCh)
|
||
drops := 0
|
||
for {
|
||
msg, err := stream.Recv()
|
||
if err != nil {
|
||
errCh <- err
|
||
return
|
||
}
|
||
select {
|
||
case inboundCh <- msg:
|
||
case <-time.After(10 * time.Millisecond):
|
||
drops++
|
||
if drops > 1000 { // 连续丢弃过多，通知主循环终止
|
||
errCh <- status.Errorf(codes.ResourceExhausted,
|
||
"too many messages dropped")
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
|
||
// 主循环同上，新增对 errCh 的监听即可
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case err := <-errCh:
|
||
return err
|
||
case envelope := <-inboundCh:
|
||
if err := stream.Send(envelope); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
> [!tip] bidirectional streaming 的设计直觉
|
||
> 想象成一个管道：一端进水（Recv）、一端出水（Send）。中间的业务逻辑可以是任意的转换、过滤、聚合。但管壁（context）不能漏。
|
||
|
||
## 错误处理对照表
|
||
|
||
streaming handler 中的错误来源复杂，不同错误的处理方式也不同：
|
||
|
||
| 错误来源 | stream.Recv 返回值 | 处理方式 |
|
||
|----------|-------------------|---------|
|
||
| Client 正常关闭连接 | `io.EOF` | `break` 或 `return nil` |
|
||
| Client 主动取消请求 | `context.Canceled`（包装为普通 error） | 记录日志，`break` |
|
||
| Network 中断 | 非 nil error（含 canceled） | 记录日志，`break` |
|
||
| 数据格式错误 | 具体 error | 原样返回给 client |
|
||
| 业务校验失败 | `status.Errorf(codes.InvalidArgument, ...)` | 返回具体错误码，终止循环 |
|
||
| 数据量超限 | `status.Errorf(codes.ResourceExhausted, ...)` | 返回错误码，终止循环 |
|
||
|
||
> [!note] 关于 Recv 返回 error 的细节
|
||
> gRPC-go 会把 `context.Canceled` 包装成普通 error 返回，不会返回 `io.EOF`。所以判断 client 断开应该同时检查两种 case。
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
A["Recv() returns error?"] -->|"No"| B["继续处理消息"]
|
||
A -->|"Yes"| C{"err == io.EOF?"}
|
||
C -->|"Yes"| D["client 正常关闭<br/>return nil ✅"]
|
||
C -->|"No"| E{"codes.FromError is OK?"}
|
||
E -->|"Yes"| F["canceled / network issue<br/>return ctx.Err() ⚠️"]
|
||
E -->|"No"| G["真实错误<br/>return status.Errorf ❌"]
|
||
```
|
||
|
||
## Context 超时传播
|
||
|
||
Streaming handler 中，context 超时由 client 侧的 WithTimeout 控制，server 必须主动感知并退出：
|
||
|
||
```go
|
||
func (s *server) LongRunningOp(req *pb.LongRequest, stream pb.Service_LongRunningOpServer) error {
|
||
ctx := stream.Context()
|
||
|
||
for i := 0; i < 100; i++ {
|
||
// 每一步都检查 context
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err() // client 取消了
|
||
default:
|
||
}
|
||
|
||
result := doStep(i)
|
||
if err := stream.Send(result); err != nil {
|
||
return err // send 失败也立即退出
|
||
}
|
||
time.Sleep(100 * time.Millisecond)
|
||
}
|
||
return nil
|
||
}
|
||
```
|
||
|
||
这个模式的精髓是 **"每一次 IO 操作前都检查 context"**。只要有一次遗漏，就可能变成孤儿 goroutine。
|
||
|
||
## 关闭与连接检测
|
||
|
||
在 streaming 场景中，理解"如何感知对端断开"比"如何发消息"更重要。
|
||
|
||
### Client 关闭的信号路径
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
A["Client 正常关闭发送端"] -->|"Recv() → io.EOF"| B["handler return nil ✅"]
|
||
C["Client Cancel context"] -->|"Recv() → ctx.Err()"| D["handler return ctx.Err() ⚠️"]
|
||
E["网络断开"] -->|"Recv() → connection error"| F["handler return error ⚠️"]
|
||
G["不读 response"] -->|"Send() → stream error"| H["handler return error ❌"]
|
||
```
|
||
|
||
| Client 行为 | Server 感知现象 | handler 处理 |
|
||
|-------------|----------------|-------------|
|
||
| 正常关闭发送端 | `Recv()` 返回 `io.EOF` | `return nil` ✅ |
|
||
| Cancel context | `Recv()` 返回 context canceled | `return ctx.Err()` ⚠️ |
|
||
| 网络断开 | `Recv()` 返回连接相关 error | `return err` ⚠️ |
|
||
| 不读 response | `Send()` 返回 stream error | `return err` ❌ |
|
||
|
||
**核心原则**：无论哪种情况，handler 的正确做法都是尽快 return，让 gRPC 内部清理资源。不需要手动 close channel 或做其他清理。
|
||
|
||
### 为什么不存在 CloseSend 通知 channel
|
||
|
||
有些开发者误以为 gRPC 提供了类似 `CloseSend() chan struct{}` 的通知通道。实际上 **gRPC-Go 并没有这样的 API**。Server 只能通过以下两种方式检测 client 停止发送：
|
||
|
||
1. **`Recv() io.EOF`** —— client 正常调用 `CloseSend()` 后
|
||
2. **`Recv() error`** —— context 取消或网络异常
|
||
|
||
任何依赖"某个 channel 被关闭"来判断的模式都是不可靠的。
|
||
|
||
## 资源清理模式
|
||
|
||
Streaming handler 中经常需要订阅外部事件源（DB watcher、消息队列等），资源清理是关键：
|
||
|
||
```go
|
||
func (s *server) WatchEvents(req *pb.WatchRequest, stream pb.Service_WatchServer) error {
|
||
ctx := stream.Context()
|
||
|
||
// 订阅事件流 —— 必须传 context
|
||
ch, cancel := s.eventBroker.Subscribe(ctx, req.GetFilter())
|
||
defer cancel() // handler 返回时自动释放订阅
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return nil // client 断了
|
||
case event, ok := <-ch:
|
||
if !ok {
|
||
return nil // broker 关闭了 channel
|
||
}
|
||
if err := stream.Send(event); err != nil {
|
||
return err // send 失败
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
要点总结：
|
||
|
||
1. `Subscribe(ctx, ...)` 一定要传 context，不能用 `context.Background()`
|
||
2. `defer cancel()` 确保 handler 退出时释放订阅
|
||
3. select 中始终优先检测 `ctx.Done()`
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant C as Client
|
||
participant H as Handler
|
||
participant EB as Event Broker
|
||
participant S as Stream
|
||
|
||
H->>EB: Subscribe(ctx, filter)
|
||
EB-->>H: ch, cancel
|
||
|
||
loop Until Disconnected
|
||
alt Client Still Connected
|
||
H->>H: select { ctx.Done(), <-ch }
|
||
EB->>H: event
|
||
H->>S: Send(event)
|
||
else Client Disconnected
|
||
H->>H: ctx.Done() triggers
|
||
H->>H: defer cancel()
|
||
H->>EB: Unsubscribe via cancel
|
||
H-->>C: Return nil
|
||
end
|
||
end
|
||
```
|
||
|
||
## 生产实践要点
|
||
|
||
写完 handler 只是第一步，生产环境还需要考虑以下方面：
|
||
|
||
### 背压与速率限制
|
||
|
||
- **Server Streaming** 中如果 client 读取速度慢于 server 发送速度，gRPC 会在 TCP/HTTP2 层积压数据。建议：监控 Send 耗时，如果 P99 持续升高则降低发送频率或加入 rate limiter
|
||
- **Client Streaming** 中可通过 `maxReceiveMessageSize` dial option 限制最大消息大小
|
||
|
||
### 监控指标
|
||
|
||
在 streaming handler 中埋点三个关键指标：
|
||
|
||
```go
|
||
// 1. handler 存活时长
|
||
duration := time.Since(start)
|
||
histogram.Record("grpc.stream.duration", duration.Milliseconds())
|
||
|
||
// 2. 发送/接收消息计数
|
||
counter.Increment("grpc.stream.messages_sent")
|
||
|
||
// 3. 非正常终止次数
|
||
counter.Increment("grpc.stream.errors", errorCode)
|
||
```
|
||
|
||
### 优雅关闭（Graceful Stop）
|
||
|
||
当 server 要 shutdown 时，不能直接 kill 正在执行的 streaming handler。正确流程：
|
||
|
||
1. 调用 `grpc.Server.GracefulStop()` —— 拒绝新连接
|
||
2. 等待已有 handler 自然返回（可以设 deadline）
|
||
3. 超时后强制 shutdown
|
||
|
||
```go
|
||
srv.GracefulStop() // 停止接收新请求
|
||
// 或带超时的优雅关闭：
|
||
stopCh := make(chan struct{})
|
||
go func() {
|
||
srv.GracefulStop()
|
||
close(stopCh)
|
||
}()
|
||
select {
|
||
case <-stopCh:
|
||
log.Println("all handlers finished")
|
||
case <-time.After(30 * time.Second):
|
||
srv.Stop() // 强制杀
|
||
log.Println("force stopped after timeout")
|
||
}
|
||
```
|
||
|
||
## 常见问题 Checklist
|
||
|
||
写完 streaming handler 后逐项核对：
|
||
|
||
- [ ] **Recv 错误分清了 `io.EOF` 和 `ctx.Err()` 吗？**（正常结束 vs client 取消的处理不同）
|
||
- [ ] Context 取消了吗？（`ctx.Done()` 在所有关键路径被检测）
|
||
- [ ] Send 失败正确处理了吗？（不会因为 send 错误继续无效循环）
|
||
- [ ] goroutine 会泄漏吗？（所有 spawn 的 goroutine 都有明确退出条件，bidirectional 中 recv goroutine 的错误能传播到主循环）
|
||
- [ ] Channel 有 buffer 吗？（bidirectional 模式下 sender/receiver 互相不饿死）
|
||
- [ ] 内存有上限吗？（client streaming 中对累积的消息数量做了限制）
|
||
- [ ] **消息大小有限制吗？**（gRPC 默认 4MB 限制，超出会直接报错）
|
||
- [ ] Subscribe 传了 context 吗？（外部事件源随 handler 一起退出）
|
||
- [ ] 大数据集做了间隔检测吗？（百万级数据不会无意义地每个元素都 select）
|
||
- [ ] 背压策略选对了吗？（drop/block/backpressure/reject 符合业务需求）
|
||
|
||
## 关联笔记
|
||
|
||
- [[hhs/gRPC/3. 服务端实现/08-Server 搭建与注册]]
|
||
- [[hhs/gRPC/3. 服务端实现/10-健康检查与反射]]
|
||
- [[hhs/gRPC/4. 客户端开发/13-Streaming Client]]
|
||
- [[hhs/gRPC/6. 工程实践篇/20-性能优化与压测]]
|