RPC 与 gRPC：像调用函数一样使用，也要按网络请求处理

本地函数返回错误时，调用方通常知道函数已经执行到哪一步。远程调用超时则不同：服务端可能没收到，也可能已经完成，只是结果没传回来。RPC 把编码、发包和分发藏在生成代码后面，却没有消除网络的不确定性。

gRPC 适合需要强契约、多语言客户端和流式通信的服务。要用好它，先分清三层：Protocol Buffers 描述和编码消息，gRPC 定义调用、状态和流，业务自己负责权限、幂等与事务。本篇使用固定依赖 grpc-go v1.75.1、protobuf-go v1.36.6，在 Go 1.27.2、Linux/amd64 上实际测试四种调用模式；这些固定版本是复现基线，不是“当前最新版本”声明。

从 proto 到一次调用，各层做什么

正在绘制图示…

查看图源代码

flowchart LR
    A[.proto 消息和服务定义] --> B[代码生成器]
    B --> C[客户端 stub]
    B --> D[服务端接口与注册代码]
    C --> E[ClientConn：连接、解析与负载均衡]
    E --> F[gRPC 消息与 HTTP/2 流]
    F --> G[服务端拦截器]
    G --> D
    D --> H[业务实现]

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码

flowchart LR
    A[.proto 消息和服务定义] --> B[代码生成器]
    B --> C[客户端 stub]
    B --> D[服务端接口与注册代码]
    C --> E[ClientConn：连接、解析与负载均衡]
    E --> F[gRPC 消息与 HTTP/2 流]
    F --> G[服务端拦截器]
    G --> D
    D --> H[业务实现]

层	常用入口	不替你解决什么
Protobuf 消息	`.proto`、`proto.Marshal/Unmarshal`	字段业务合法性、访问权限
生成代码	`NewXClient`、`RegisterXServer`	网络一定成功、业务重试安全
grpc-go 客户端	`grpc.NewClient`、调用时的 Context	自动给每次调用合理 deadline
grpc-go 服务端	`grpc.NewServer`、`Serve`	所有方法的认证、业务状态一致性
应用服务	实现生成接口	仍需处理部分失败、持久化和副作用

ClientConn 是长期复用的逻辑连接，不要每次 RPC 创建并关闭。它可以管理底层连接与重连；grpc.NewClient 创建时不等于已经完成一次网络连接或健康检查，实际通信按调用需要启动。测试用 bufconn 提供进程内连接，仍经过 gRPC 编解码和流处理，不是直接调用服务函数。grpc.NewClient v1.75.1

示例的 insecure.NewCredentials() 仅用于这条进程内测试链路。生产网络要配置合适的 TLS、服务身份与访问控制，不能照抄成公网服务。

Protobuf 的字段号是协议的一部分

proto

message Item {
  string id = 1;
  optional int32 count = 2;
}

1、2 是字段号，二进制报文按字段号识别数据，不会把 id 这个字段名每次都传一遍。类型决定编码方式。这里给 count 加 optional，是为了区分“未提供”与“明确提供 0”；proto3 的普通隐式存在标量无法保留这个区分。Protobuf 字段存在性

用官方已生成的 HealthCheckRequest 做最小实验：

b, err := proto.Marshal(&healthpb.HealthCheckRequest{Service: "svc"})
if err != nil { panic(err) }
fmt.Printf("%x\n", b)
var got healthpb.HealthCheckRequest
if err := proto.Unmarshal(b, &got); err != nil { panic(err) }
fmt.Println(got.GetService())

实际输出：

text

0a03737663
svc

0a 表示字段 1、长度限定类型，03 是后面字符串的字节长度，剩下三个字节是 svc。这只是这个小消息的编码解释，不能据此假设任意消息有唯一的字节排列，更不能把默认序列化结果直接当跨版本稳定签名格式。Protobuf 编码、序列化不是规范化编码

演进时先看旧程序会怎样理解

修改	通常的处理原则	为什么
增加新字段	使用新字段号，给缺失情况合理语义	旧端不知道新字段，新端会遇到旧消息
删除字段	保留其编号和名称为 reserved	避免以后把旧数据误认成新含义
改字段号	视为删除旧字段并新增字段	二进制身份已经改变
改字段类型	按官方兼容表逐项审查	线格式可解析也可能丢值或改语义
增加枚举值	客户端必须能处理未知值	穷举分支不一定有回退
改 JSON 字段名	另审 ProtoJSON 兼容性	二进制按编号，JSON 按名称

“二进制兼容”不等于业务兼容。把 count 单位从“个”改成“箱”，编码完全没变，业务已经不兼容。经 JSON 中转还可能丢失未知字段，因此不能只测二进制往返。proto3 更新规则、ProtoJSON 格式与兼容性

四种调用模式，差别在两边何时结束发送

模式	IDL 形状	调用方常用动作	适合场景
一元	`Get(Item) returns (Item)`	发一个请求，拿结果或错误	查询、命令
服务端流	`List(Query) returns (stream Item)`	一次请求，循环 Recv	列表、订阅、下载
客户端流	`Upload(stream Item) returns (Summary)`	多次 Send，CloseAndRecv	分批上传、汇总
双向流	`Exchange(stream Item) returns (stream Item)`	Send 与 Recv 独立推进	交互、持续状态交换

流内的消息有顺序；双向两条流之间没有天然的“一发一回”配对。要配对，需要应用消息里的 request ID。文末复用 grpc-go 官方生成的测试服务类型来验证三种流，另用健康服务验证一元；这样读者不需要安装 protoc 也能重跑。四模式的示意 IDL 单独附在末尾，未在本次生成代码。gRPC Go 基础教程

一元：error 也是协议结果

普通成功调用返回 response 和 nil；出错时检查 status.Code(err)，不要把错误文字当稳定机器标识。本次健康服务要求测试 Metadata，并检查收到的 Context 有 deadline：没有 Metadata 返回 Unauthenticated；带 Metadata 返回 SERVING；等待超时返回 DeadlineExceeded，均由测试断言。

ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
ctx = metadata.AppendToOutgoingContext(ctx, "x-demo", "test")
resp, err := client.Check(ctx, &healthpb.HealthCheckRequest{})

3 秒是测试预算，不是通用建议值。实际应从调用链的用户体验与下游延迟预算推导。

服务端流：读到消息以后还要读最终状态

接收端循环的关键是分别处理 EOF 和其他错误：

for {
    message, err := stream.Recv()
    if err == io.EOF { break }
    if err != nil { return err }
    consume(message)
}

这里 consume 是业务处理的占位动作；文末测试收集 Payload。正常流的实测输出：

text

server stream: [a b] then EOF

另一个请求让服务端先 Send a，再返回 Internal：

text

partial stream: a, then Internal

所以“拿到了第一条消息”不能等同于“RPC 成功”。最后的非 EOF 错误必须处理；若业务要求全量成功后才生效，需要暂存或分批确认策略。已经消费的消息不会因为最终失败自动撤销。

客户端流：CloseAndRecv 表示发送完，再取汇总

示例发送 ab 与 cde 两块。服务端循环 Recv，累加长度；遇到输入 EOF 后 SendAndClose 一次返回总量。客户端的关键两步：

// 每一次 Send 都检查错误；消息类型与构造见文末。
err := input.Send(part)
if err != nil { return err }
summary, err := input.CloseAndRecv()

测试完整发送两块后的输出：

text

client stream total: 5

流式并不等于无限大。每个消息、整次上传、持续时间都要限制；累加器也要选择适合业务范围的类型并检查溢出。本例只使用几个字节，不能当成不限量上传服务。

双向流：CloseSend 只关闭自己发送的一半

测试先发 hello，服务端立即回显；客户端拿到回显后调用 CloseSend，服务端读到 EOF 后结束，客户端再 Recv 得到 EOF：

text

bidi: hello, half-close, EOF

如果双方都先等对方发，通信就停住；如果双方都只写不读，又可能被流控卡住。常见结构是一条发送路径、一条接收路径，共享取消信号和有限缓冲。grpc-go 允许同一流的一次发送和一次接收并行，但不要让多个 goroutine 同时 Send，或同时 Recv；发送后也不要修改正在被框架使用的消息。grpc-go v1.75.1 stream 接口约定

背压让慢接收方能影响发送方

gRPC 的流控会限制尚未被接收方消费的数据。Send 可能等待；返回成功也不表示对方业务已经处理。若你在 Send 前面放一个无限增长的队列，就绕开了这种限制，把压力变成了自己的内存增长。gRPC 流控

正在绘制图示…

查看图源代码

flowchart LR
    A[业务生产消息] --> B[有界队列]
    B --> C[Send]
    C --> D[传输流控]
    D --> E[接收端 Recv]
    E --> F[业务处理]
    F -. 消费变慢 .-> E
    E -. 缓冲与窗口逐渐耗尽 .-> D
    D -. Send 可能等待 .-> C
    C -. 生产端等待或按策略拒绝 .-> B

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码

flowchart LR
    A[业务生产消息] --> B[有界队列]
    B --> C[Send]
    C --> D[传输流控]
    D --> E[接收端 Recv]
    E --> F[业务处理]
    F -. 消费变慢 .-> E
    E -. 缓冲与窗口逐渐耗尽 .-> D
    D -. Send 可能等待 .-> C
    C -. 生产端等待或按策略拒绝 .-> B

有些业务可以丢弃中间状态，只保留最新位置；有些必须逐条落库确认。背压策略属于业务契约，不能只凭“用了 streaming”决定。

拦截器和 Metadata：统一入口，但分一元与流

一元拦截器接收 Context、请求、方法信息与 next；流拦截器接收 ServerStream 和 next。它们可以做认证、日志、度量和 Context 注入，但两类要分别注册。只安装 UnaryInterceptor 不会自动保护 Streaming RPC。

文末流拦截器从 ss.Context() 读取测试 Metadata，缺少值就返回 Unauthenticated；通过后才调用 next。测试还特意在客户端第一次 Recv 时观察拒绝，因为创建流成功返回，不代表服务端已接受调用。

md, _ := metadata.FromIncomingContext(ss.Context())
if values := md.Get("x-demo"); len(values) != 1 || values[0] != "test" {
    return status.Error(codes.Unauthenticated, "missing metadata")
}
return next(srv, ss)

多个拦截器按注册顺序嵌套，最前面的位于外层。流级拦截器不是自动的“每消息回调”；若要观察每次 SendMsg/RecvMsg，需要包装 ServerStream。改流的 Context 同样要包装其 Context() 方法。ChainUnaryInterceptor、StreamServerInterceptor

Metadata 是附带键值元信息，键按小写处理，二进制值用 -bin 后缀；grpc- 前缀保留给框架。不要把大型业务数据塞进去。入站 Metadata 和出站 Metadata 也不是一回事，转发下游时应明确选取需要的追踪、身份信息，不能无条件传播所有客户端 Header。gRPC Metadata

Deadline 是剩余预算，不能每一跳重新充值

客户端默认未必有符合业务要求的 deadline，应该显式设置。服务端接到请求后，要用入站 ctx 派生下游调用；如果换成 context.Background()，调用链的取消与剩余预算就丢了。

child, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
defer cancel()
return downstream.Check(child, request)

若入站只剩 80ms，子 Context 不会延长到 200ms；它最多还能用父级剩余预算。Go gRPC 能把传入 Context 的 deadline 传播到远端，但前提是应用把正确的 Context 交给调用，跨 HTTP、消息队列或其他语言组件时也要核对其规则。所谓“自动传播”，不意味着在任意新建 Context、任意异步任务里都自动保持。gRPC Deadlines

超时之后，服务端应检查 Context 并停止工作；框架不能强杀不理会取消的业务代码。更重要的是，DeadlineExceeded 不证明服务端没有提交副作用。创建订单、扣库存等操作仍要有业务幂等键。

错误码决定下一步，不能照搬 HTTP 数字

gRPC code	常见含义	调用方通常怎么处理
InvalidArgument	输入本身错误	修改输入
Unauthenticated	无法确认身份	重新获取或修正凭据
PermissionDenied	已知主体无权操作	检查授权，不盲目重试
NotFound / AlreadyExists	对象状态与操作不符	按业务状态处理
FailedPrecondition	需要先满足业务前提	改变状态再尝试
Aborted	并发或事务冲突	在正确层级重做事务
ResourceExhausted	配额或资源不足	退避、减载、检查配额
Unavailable	暂时无法提供服务	只在操作可重试且有预算时重试
DeadlineExceeded / Canceled	超时 / 取消	结果可能未知，先考虑幂等与查询
Internal / Unknown	内部问题 / 未分类错误	记录关联信息，避免暴露细节

这些是 gRPC 状态，不是 HTTP 状态码。正常的 HTTP/2 响应头可能是 200，但 gRPC 最终状态仍然是错误；需要 gRPC 客户端读取状态与 Trailer。gRPC 状态码、HTTP/2 上的 gRPC 协议

配置重试要限制尝试次数、总 deadline、退避和抖动，并区分框架透明重试与服务配置的策略重试。响应头到达后，调用通常已被视为 committed，不再按重试策略从头重放；流已经向应用交付消息时，更不能把“断了重连”当成透明续传。业务是否允许重复执行、怎样恢复位置，必须另行定义。gRPC Retry

grpc-gateway：把 HTTP/JSON 映射到 gRPC 契约

grpc-gateway 根据 proto 注解或外部配置生成反向代理，让例如 GET /v1/items/42 映射成某个 gRPC 方法。它适合已有 gRPC 服务又需要 HTTP/JSON 接口的场景；不是把所有 gRPC 能力原封不动搬进浏览器。

映射位置	要明确的决定
HTTP 路径、Query、Body → protobuf	参数放哪，缺失值如何解释
HTTP Header → Metadata	哪些可以转发，哪些必须丢弃
gRPC status → HTTP 错误	对外状态码、统一错误体和敏感字段
protobuf → JSON	使用 ProtoJSON 映射，int64、枚举、字段名等与普通 encoding/json 不同
服务端流 → HTTP 流	分帧格式、部分成功、流内错误、代理缓冲

Gateway 的 Unary 错误处理与流内错误处理有不同扩展入口，因为流发送后 HTTP 状态已经提交。客户端流和双向流也不能假定普通 HTTP/JSON 代理原生提供同等交互能力；先核对所用版本与客户端支持，再选 gRPC-Web、Connect 或独立协议。grpc-gateway 定制说明、项目功能边界

本次没有生成或启动 gateway，也没有运行 protoc。可运行实验使用固定模块内已生成的官方消息与服务代码。这样既能验证 RPC 行为，也不把未执行的代码生成过程写成亲测。

资料与验证范围

固定 grpc-go v1.75.1 API、stream.go、官方测试服务生成代码、健康服务生成代码。
protobuf-go v1.36.6 proto API 与官方 proto3、字段存在性、ProtoJSON 文档。
gRPC 官方 Go 基础、流控、Metadata、deadline、重试、状态码。
grpc-gateway 官方映射定制。

go mod verify、go test、go test -race、go vet 均通过。测试覆盖一元/三种流、Metadata 拒绝、两类拦截器、入站 deadline、超时、EOF、半关闭、收到部分消息后最终失败，以及 protobuf 编解码。未测真实 TLS、跨机网络、负载均衡、生产流控压力或 gateway。

完整复现代码与依赖

使用 Go 1.27.2。首次需要获取固定依赖时，可使用官方 Go 代理与校验服务；缓存齐全后测试本身可以离线：

GOPROXY=https://proxy.golang.org GOSUMDB=sum.golang.org go mod download
go mod verify
GOPROXY=off GOSUMDB=off go test -v -count=1 ./...
GOPROXY=off GOSUMDB=off go test -race -count=1 ./...
GOPROXY=off GOSUMDB=off go vet ./...

modes.proto 只是前面四种模式的完整示意契约；测试不依赖它生成的文件。真正用到的生成类型来自 go.mod 固定的官方模块。不要把本文测试服务作为业务 API 依赖。

go.mod

module example.com/webapi/05_grpc

go 1.27.2

require google.golang.org/grpc v1.75.1

require (
	golang.org/x/net v0.41.0 // indirect
	golang.org/x/sys v0.33.0 // indirect
	golang.org/x/text v0.26.0 // indirect
	google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 // indirect
	google.golang.org/protobuf v1.36.6 // indirect
)

go.sum

代码块 · 36 行 · 点击收起 / 展开

text

github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI=
go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg=
go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc=
go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps=
go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw=
golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA=
golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
golang.org/x/text v0.26.0 h1:P42AVeLghgTYr4+xUnTRKDMqpar+PtX7KWuNQL21L8M=
golang.org/x/text v0.26.0/go.mod h1:QK15LZJUUQVJxhz7wXgxSy/CJaTFjd0G+YLonydOVQA=
gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 h1:pFyd6EwwL2TqFf8emdthzeX+gZE1ElRq3iM8pui4KBY=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI=
google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=

grpc_test.go

代码块 · 88 行 · 点击收起 / 展开

package grpcdemo

import (
	"context"
	"io"
	"net"
	"testing"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	pb "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
)

type service struct{ pb.UnimplementedHealthServer }

func (service) Check(ctx context.Context, r *pb.HealthCheckRequest) (*pb.HealthCheckResponse, error) {
	if _, ok := ctx.Deadline(); !ok {
		return nil, status.Error(codes.InvalidArgument, "deadline required")
	}
	if r.Service == "wait" {
		<-ctx.Done()
		return nil, status.FromContextError(ctx.Err()).Err()
	}
	return &pb.HealthCheckResponse{Status: pb.HealthCheckResponse_SERVING}, nil
}
func (service) Watch(r *pb.HealthCheckRequest, s pb.Health_WatchServer) error {
	for _, v := range []pb.HealthCheckResponse_ServingStatus{pb.HealthCheckResponse_SERVING, pb.HealthCheckResponse_NOT_SERVING} {
		if err := s.Send(&pb.HealthCheckResponse{Status: v}); err != nil {
			return err
		}
	}
	return nil
}
func unary(ctx context.Context, req any, info *grpc.UnaryServerInfo, next grpc.UnaryHandler) (any, error) {
	md, _ := metadata.FromIncomingContext(ctx)
	if v := md.Get("x-demo"); len(v) != 1 || v[0] != "test" {
		return nil, status.Error(codes.Unauthenticated, "test metadata missing")
	}
	return next(ctx, req)
}
func TestDeadlineMetadataAndStream(t *testing.T) {
	lis := bufconn.Listen(1 << 20)
	srv := grpc.NewServer(grpc.UnaryInterceptor(unary))
	pb.RegisterHealthServer(srv, service{})
	done := make(chan error, 1)
	go func() { done <- srv.Serve(lis) }()
	defer func() { srv.Stop(); lis.Close(); <-done }()
	cc, err := grpc.NewClient("passthrough:///local", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }))
	if err != nil {
		t.Fatal(err)
	}
	defer cc.Close()
	client := pb.NewHealthClient(cc)
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	if _, err = client.Check(ctx, &pb.HealthCheckRequest{}); status.Code(err) != codes.Unauthenticated {
		t.Fatalf("code=%v err=%v", status.Code(err), err)
	}
	ctx = metadata.AppendToOutgoingContext(ctx, "x-demo", "test")
	resp, err := client.Check(ctx, &pb.HealthCheckRequest{})
	if err != nil || resp.GetStatus() != pb.HealthCheckResponse_SERVING {
		t.Fatalf("%v %v", resp, err)
	}
	stream, err := client.Watch(ctx, &pb.HealthCheckRequest{})
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []pb.HealthCheckResponse_ServingStatus{pb.HealthCheckResponse_SERVING, pb.HealthCheckResponse_NOT_SERVING} {
		got, e := stream.Recv()
		if e != nil || got.GetStatus() != want {
			t.Fatalf("%v %v", got, e)
		}
	}
	if _, err = stream.Recv(); err != io.EOF {
		t.Fatalf("stream terminal=%v", err)
	}
	short, stop := context.WithTimeout(ctx, 30*time.Millisecond)
	defer stop()
	if _, err = client.Check(short, &pb.HealthCheckRequest{Service: "wait"}); status.Code(err) != codes.DeadlineExceeded {
		t.Fatalf("deadline error=%v", err)
	}
	t.Log("unary metadata/interceptor, incoming deadline, server stream ordering and EOF, DeadlineExceeded verified")
}

stream_test.go

代码块 · 179 行 · 点击收起 / 展开

package grpcdemo

import (
	"context"
	"fmt"
	"io"
	"net"
	"sync/atomic"
	"testing"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	pb "google.golang.org/grpc/interop/grpc_testing"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
	"google.golang.org/protobuf/proto"
)

type streams struct {
	pb.UnimplementedTestServiceServer
}

func (streams) StreamingOutputCall(r *pb.StreamingOutputCallRequest, s grpc.ServerStreamingServer[pb.StreamingOutputCallResponse]) error {
	for i, body := range []string{"a", "b"} {
		if err := s.Send(&pb.StreamingOutputCallResponse{Payload: &pb.Payload{Body: []byte(body)}}); err != nil {
			return err
		}
		if i == 0 && r.ResponseStatus != nil {
			return status.Error(codes.Code(r.ResponseStatus.Code), r.ResponseStatus.Message)
		}
	}
	return nil
}
func (streams) StreamingInputCall(s grpc.ClientStreamingServer[pb.StreamingInputCallRequest, pb.StreamingInputCallResponse]) error {
	var total int32
	for {
		r, err := s.Recv()
		if err == io.EOF {
			return s.SendAndClose(&pb.StreamingInputCallResponse{AggregatedPayloadSize: total})
		}
		if err != nil {
			return err
		}
		total += int32(len(r.GetPayload().GetBody()))
	}
}
func (streams) FullDuplexCall(s grpc.BidiStreamingServer[pb.StreamingOutputCallRequest, pb.StreamingOutputCallResponse]) error {
	for {
		r, err := s.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		if err = s.Send(&pb.StreamingOutputCallResponse{Payload: r.Payload}); err != nil {
			return err
		}
	}
}
func TestStreamingModesAndPartialError(t *testing.T) {
	lis := bufconn.Listen(1 << 20)
	var intercepted atomic.Int32
	interceptor := func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, next grpc.StreamHandler) error {
		md, _ := metadata.FromIncomingContext(ss.Context())
		if len(md.Get("x-demo")) != 1 || md.Get("x-demo")[0] != "test" {
			return status.Error(codes.Unauthenticated, "missing metadata")
		}
		intercepted.Add(1)
		return next(srv, ss)
	}
	server := grpc.NewServer(grpc.StreamInterceptor(interceptor))
	pb.RegisterTestServiceServer(server, streams{})
	done := make(chan error, 1)
	go func() { done <- server.Serve(lis) }()
	defer func() { server.Stop(); lis.Close(); <-done }()
	conn, err := grpc.NewClient("passthrough:///local", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }))
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	c := pb.NewTestServiceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	denied, err := c.StreamingOutputCall(ctx, &pb.StreamingOutputCallRequest{})
	if err == nil {
		_, err = denied.Recv()
	}
	if status.Code(err) != codes.Unauthenticated {
		t.Fatal(err)
	}
	ctx = metadata.AppendToOutgoingContext(ctx, "x-demo", "test")
	output, err := c.StreamingOutputCall(ctx, &pb.StreamingOutputCallRequest{})
	if err != nil {
		t.Fatal(err)
	}
	var values []string
	for {
		m, e := output.Recv()
		if e == io.EOF {
			break
		}
		if e != nil {
			t.Fatal(e)
		}
		values = append(values, string(m.GetPayload().GetBody()))
	}
	if fmt.Sprint(values) != "[a b]" {
		t.Fatal(values)
	}
	t.Log("server stream:", values, "then EOF")
	input, err := c.StreamingInputCall(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, body := range []string{"ab", "cde"} {
		if err := input.Send(&pb.StreamingInputCallRequest{Payload: &pb.Payload{Body: []byte(body)}}); err != nil {
			t.Fatal(err)
		}
	}
	total, err := input.CloseAndRecv()
	if err != nil || total.GetAggregatedPayloadSize() != 5 {
		t.Fatalf("%v %v", total, err)
	}
	t.Log("client stream total:", total.GetAggregatedPayloadSize())
	bidi, err := c.FullDuplexCall(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err = bidi.Send(&pb.StreamingOutputCallRequest{Payload: &pb.Payload{Body: []byte("hello")}}); err != nil {
		t.Fatal(err)
	}
	reply, err := bidi.Recv()
	if err != nil || string(reply.GetPayload().GetBody()) != "hello" {
		t.Fatalf("%v %v", reply, err)
	}
	if err = bidi.CloseSend(); err != nil {
		t.Fatal(err)
	}
	if _, err = bidi.Recv(); err != io.EOF {
		t.Fatal(err)
	}
	t.Log("bidi: hello, half-close, EOF")
	partial, err := c.StreamingOutputCall(ctx, &pb.StreamingOutputCallRequest{ResponseStatus: &pb.EchoStatus{Code: int32(codes.Internal), Message: "demo failure"}})
	if err != nil {
		t.Fatal(err)
	}
	first, err := partial.Recv()
	if err != nil || string(first.GetPayload().GetBody()) != "a" {
		t.Fatal(first, err)
	}
	_, err = partial.Recv()
	if status.Code(err) != codes.Internal {
		t.Fatal(err)
	}
	t.Log("partial stream: a, then", status.Code(err))
	if intercepted.Load() != 4 {
		t.Fatal(intercepted.Load())
	}
}
func Example_protobufWire() {
	b, err := proto.Marshal(&healthpb.HealthCheckRequest{Service: "svc"})
	if err != nil {
		panic(err)
	}
	fmt.Printf("%x\n", b)
	var got healthpb.HealthCheckRequest
	if err = proto.Unmarshal(b, &got); err != nil {
		panic(err)
	}
	fmt.Println(got.GetService())
	// Output:
	// 0a03737663
	// svc
}

modes.proto

proto

syntax = "proto3";
package teaching.v1;
option go_package = "example.com/teaching/gen/teaching/v1;teachingv1";
message Item { string id = 1; optional int32 count = 2; }
message ListRequest { string prefix = 1; }
message Summary { int64 count = 1; }
service Items {
  rpc Get(Item) returns (Item);
  rpc List(ListRequest) returns (stream Item);
  rpc Upload(stream Item) returns (Summary);
  rpc Exchange(stream Item) returns (stream Item);
}

RPC 与 gRPC：像调用函数一样使用，也要按网络请求处理 ​

从 proto 到一次调用，各层做什么 ​

Protobuf 的字段号是协议的一部分 ​

演进时先看旧程序会怎样理解 ​

四种调用模式，差别在两边何时结束发送 ​

一元：error 也是协议结果 ​

服务端流：读到消息以后还要读最终状态 ​

客户端流：CloseAndRecv 表示发送完，再取汇总 ​

双向流：CloseSend 只关闭自己发送的一半 ​

背压让慢接收方能影响发送方 ​

拦截器和 Metadata：统一入口，但分一元与流 ​

Deadline 是剩余预算，不能每一跳重新充值 ​

错误码决定下一步，不能照搬 HTTP 数字 ​

grpc-gateway：把 HTTP/JSON 映射到 gRPC 契约 ​

资料与验证范围 ​

完整复现代码与依赖 ​

go.mod ​

go.sum ​

grpc_test.go ​

stream_test.go ​

modes.proto ​

RPC 与 gRPC：像调用函数一样使用，也要按网络请求处理

从 proto 到一次调用，各层做什么

Protobuf 的字段号是协议的一部分

演进时先看旧程序会怎样理解

四种调用模式，差别在两边何时结束发送

一元：error 也是协议结果

服务端流：读到消息以后还要读最终状态

客户端流：CloseAndRecv 表示发送完，再取汇总

双向流：CloseSend 只关闭自己发送的一半

背压让慢接收方能影响发送方

拦截器和 Metadata：统一入口，但分一元与流

Deadline 是剩余预算，不能每一跳重新充值

错误码决定下一步，不能照搬 HTTP 数字

grpc-gateway：把 HTTP/JSON 映射到 gRPC 契约

资料与验证范围

完整复现代码与依赖

go.mod

go.sum

grpc_test.go

stream_test.go

modes.proto