Skip to content

gRPC ​

gRPC 是一套基于服务定义生成客户端与服务端代码的 RPC 框架。Go 项目通常用 Protocol Buffers 描述消息和服务,调用数据通过 HTTP/2 连接传输。本文先用 net/rpc 说明 RPC 的基本过程,再进入 Protobuf、流式调用、元数据、拦截器和错误处理。

1 RPC ​

RPC(Remote Procedure Call,远程过程调用)是一种通信模型,不是某一个固定协议。调用方使用本地接口发起请求,框架负责消息编码、网络传输和结果返回。网络故障、超时和重复执行等问题并不会因此消失,业务代码仍要处理这些边界。

1.1 基本原理 ​

  • 本地调用:程序调用一个本地函数,直接执行并返回结果。
  • 远程调用:客户端把方法和参数编码后发送到服务端,服务端执行对应方法,再返回结果或错误。

1.2 工作流程 ​

  1. 客户端调用:客户端像调用本地方法一样调用远程方法。
  2. 编码:客户端把方法信息和参数编码为约定的传输格式。
  3. 网络传输:序列化后的数据通过网络发送到服务器。
  4. 解码:服务端把请求恢复为程序可用的数据结构。
  5. 执行方法:服务器根据方法名和参数执行对应的函数。
  6. 返回结果:服务端编码响应,并把响应状态和数据发回客户端。
  7. 客户端接收:客户端解码结果,或者处理超时、取消和远端错误。

1.3 net/rpc 服务端 ​

Go 标准库的 net/rpc 可以通过 TCP 或 HTTP 传输 RPC 消息,默认使用 Gob 编码。它只面向 Go 程序,并且已经进入冻结状态,适合用来理解 RPC 或处理简单的内部通信;跨语言接口通常使用 gRPC 等方案。

要开发一个 RPC 服务端,通常需要以下几个步骤:

  1. 定义服务类型:结构体的导出方法作为远程方法。
  2. 注册服务:把服务实例注册到 RPC 服务中。
  3. 启动服务:监听端口并处理连接。
go
type HelloService struct{}

func (s *HelloService) SayHello(request string, reply *string) error {
	*reply = "Hello " + request
	return nil
}

func main() {
	if err := rpc.RegisterName("HelloService", new(HelloService)); err != nil {
		log.Fatal(err)
	}

	listener, err := net.Listen("tcp", ":1234")
	if err != nil {
		log.Fatal(err)
	}

	for {
		conn, err := listener.Accept()
		if err != nil {
			log.Printf("accept: %v", err)
			continue
		}
		go rpc.ServeConn(conn)
	}
}

1.4 net/rpc 客户端 ​

  1. 连接到 RPC 服务:通过 TCP 或 HTTP 连接服务端。
  2. 调用远程方法:通过 RPC 客户端调用服务端提供的方法。
go
client, err := rpc.Dial("tcp", "127.0.0.1:1234")
if err != nil {
	log.Fatal(err)
}
defer client.Close()

var reply string
if err := client.Call("HelloService.SayHello", "zhang", &reply); err != nil {
	log.Fatal(err)
}
fmt.Println(reply) // Hello zhang

2 Protocol Buffers ​

Protocol Buffers(Protobuf)同时包含接口描述语言、二进制编码格式和代码生成工具。它可以脱离 gRPC 单独使用;gRPC 只是把 Protobuf 作为默认的服务定义与消息格式。

2.1 安装 ​

  1. 从 Protobuf Releases 安装 protoc,并确认它位于 PATH:

    sh
    protoc --version
  2. 安装 Go 消息代码和 gRPC 服务代码的生成插件:

    sh
    go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
    go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
  3. 确保 Go 的可执行文件目录位于 PATH:

    sh
    export PATH="$PATH:$(go env GOPATH)/bin"

实际项目应在构建脚本中固定工具版本,避免 @latest 更新后生成结果发生意外变化。

2.2 定义数据结构 ​

创建一个 .proto 文件,例如 msg.proto:

proto
syntax = "proto3";

package grpcdemo;

option go_package = "example.com/grpcdemo/proto";

message HelloRequest {
  string name = 1;  // 1 是编号,不是值
  uint32 age  = 2;
}

字段后的数字是线上的字段编号,不是数组下标。编号一旦发布就不应随意修改或分配给其他字段。go_package 应填写生成代码的完整 Go 导入路径,并与项目的模块路径保持一致。

2.3 生成 Go 代码 ​

使用 protoc 生成目标语言的代码:

sh
protoc \
    --go_out=. \
	--go_opt=paths=source_relative \
    --go-grpc_out=. \
	--go-grpc_opt=paths=source_relative \
	hello.proto

--go_out 生成消息类型,--go-grpc_out 生成 gRPC 客户端和服务端接口。paths=source_relative 表示输出文件与输入的 .proto 文件放在同一相对目录下。

如果 hello.proto 导入了其他的 .proto 文件,例如:

proto
import "google/api/annotations.proto";

则需要通过 --proto_path 指定导入根目录。假设文件位于 ../third_party/google/api/annotations.proto,应把 ../third_party 作为搜索路径:

sh
protoc \
    --proto_path=. \
    --proto_path=../third_party \
    --go_out=. \
    --go-grpc_out=. \
	hello.proto

2.4 标量类型 ​

下表列出常用标量类型及其生成代码中的对应类型:

Proto TypeGo TypeC++ TypePython Type
doublefloat64doublefloat
floatfloat32floatfloat
int32int32int32_tint
int64int64int64_tint
uint32uint32uint32_tint
uint64uint64uint64_tint
sint32int32int32_tint
sint64int64int64_tint
fixed32uint32uint32_tint
fixed64uint64uint64_tint
sfixed32int32int32_tint
sfixed64int64int64_tint
boolboolboolbool
stringstringstringstr
bytes[]bytestringbytes

2.5 默认值 ​

数据类型默认值说明
double0双精度浮点型
float0浮点型
int320使用变长编码。负数编码效率低下。如果字段可能为负,用 sint32 代替
int640使用变长编码。负数编码效率低下。如果字段可能为负,用 sint64 代替
uint320使用变长编码
uint640使用变长编码
sint320使用 ZigZag 编码,负数通常比 int32 更紧凑
sint640使用 ZigZag 编码,负数通常比 int64 更紧凑
fixed320始终为 4 字节,如果值大于 228,该类型比 uint32 高效
fixed640始终为 8 字节,如果值大于 256,该类型比 uint64 高效
sfixed320始终为 4 字节
sfixed640始终为 8 字节
boolfalse布尔型
string""必须是 UTF-8 编码的文本
bytesnilGo 中为 []byte,读取语义为空字节序列

消息类型字段的默认值为 nil,枚举字段取编号为 0 的值。Proto3 不允许使用 [default = ...] 声明自定义默认值;业务默认值应在应用代码中处理。需要区分“字段未设置”和“字段显式设置为零值”时,可以使用 optional:

proto
optional int32 page_size = 1;

2.6 消息嵌套 ​

可以使用其他消息类型作为字段类型。例如,要在每条 SearchResponse 消息中包含 Result 消息,可以直接在同一个 .proto 文件中定义一个 Result 消息类型,然后在 SearchResponse 中指定一个字段类型为 Result:

proto
message SearchResponse {
  repeated Result results = 1;
}

message Result {
  string          url      = 1;
  string          title    = 2;
  repeated string snippets = 3;
}

也可以直接在 SearchResponse 消息类型中直接定义和使用 Result 消息类型:

proto
message SearchResponse {
  message Result {
    string          url      = 1;
    string          title    = 2;
    repeated string snippets = 3;
  }
  repeated Result results = 1;
}

如果要在父消息类型之外重用此消息类型,需要使用 _Parent_._Type_ 进行调用:

proto
message SomeOtherMessage {
  SearchResponse.Result result = 1;
}

也可以根据需要嵌套多层消息,下面的例子中,两个名为 Inner 的嵌套类型是完全独立的,因为它们定义在不同的消息中:

proto
message Outer {       // Level 0
  message MiddleAA {  // Level 1
    message Inner {   // Level 2
      int64 ival  = 1;
      bool  booly = 2;
    }
  }
  message MiddleBB {  // Level 1
    message Inner {   // Level 2
      int32 ival  = 1;
      bool  booly = 2;
    }
  }
}

2.7 导入定义 ​

可以通过导入其他 .proto 文件来使用里面的定义:

proto
import "myproject/other_protos.proto";

默认情况下,只能使用直接导入的 .proto 文件中的定义。但是通过 import public 导入的 .proto 文件可以传递到下一个文件。例如:

proto
// new.proto
// 这里有很多定义
proto
// old.proto
import public "new.proto";
import "other.proto";

// 这里有很多定义
proto
// client.proto
import "old.proto";

// 可以使用 old.proto 和 new.proto 中的定义,但是不能使用 other.proto 的定义

2.8 枚举类型 ​

proto
enum Corpus {
  CORPUS_UNSPECIFIED = 0;
  CORPUS_UNIVERSAL   = 1;
  CORPUS_WEB         = 2;
  CORPUS_IMAGES      = 3;
  CORPUS_LOCAL       = 4;
  CORPUS_NEWS        = 5;
  CORPUS_PRODUCTS    = 6;
  CORPUS_VIDEO       = 7;
}

message SearchRequest {
  string query            = 1;
  int32  page_number      = 2;
  int32  results_per_page = 3;
  Corpus corpus           = 4;
}

字段 SearchRequest.corpus 的默认值是 CORPUS_UNSPECIFIED,因为它的编号为 0。Proto3 枚举定义的第一个值必须为 0,通常用 *_UNSPECIFIED 表示调用方没有明确选择。

2.9 Map 类型 ​

proto
message Msg {
  map<string, int32> mp = 1;
}

key 可以使用整数、bool 或字符串类型,不能使用浮点数、bytes 和枚举。value 可以使用除 map 以外的任意字段类型。

2.10 时间戳类型 ​

proto
import "google/protobuf/timestamp.proto";

message Msg {
  google.protobuf.Timestamp timestamp = 1;
}

google/protobuf/timestamp.proto 的源码如下:

proto
syntax = "proto3";

package google.protobuf;

option go_package = "github.com/golang/protobuf/ptypes/timestamp";

message Timestamp {
  int64 seconds = 1;
  int32 nanos   = 2;
}

在使用 Go 实现时,需要导入 google.golang.org/protobuf/types/known/timestamppb 包来生成时间戳。

2.11 字段演进 ​

删除字段后应保留原编号和名称,防止后续代码误用旧数据:

proto
message User {
  reserved 2, 4 to 6;
  reserved "nickname";

  string name = 1;
  string email = 3;
}

不要修改已经发布字段的编号,也不要把同一编号改成含义不同的字段。消息兼容性依赖编号,而不是字段名称。

3 gRPC ​

gRPC 根据 .proto 中的 service 生成强类型客户端和服务端接口。一次调用通常包含方法路径、消息、元数据、状态码和可选的流数据;Protobuf 负责消息编码,gRPC 负责调用语义与传输。

3.1 定义服务 ​

proto
syntax = "proto3";

package grpcdemo;

option go_package = "example.com/grpcdemo/proto";

service Greeter {
  rpc SayHello(HelloRequest) returns (HelloResponse);
}

message HelloRequest {
  string name = 1;
}

message HelloResponse {
  string msg = 1;
}

3.2 生成代码 ​

运行 protoc 命令生成 Go 代码。会生成两个文件:

  • hello.pb.go:消息定义
  • hello_grpc.pb.go:gRPC 服务端/客户端接口

示例模块可以这样初始化依赖:

sh
go mod init example.com/grpcdemo
go get google.golang.org/grpc
go get google.golang.org/protobuf

3.3 实现服务端 ​

go
type Server struct {
	proto.UnimplementedGreeterServer // 为后续新增 RPC 保留兼容性
}

func (s *Server) SayHello(_ context.Context, req *proto.HelloRequest) (*proto.HelloResponse, error) {
	return &proto.HelloResponse{Msg: "Hello " + req.Name}, nil
}

func main() {
	lis, err := net.Listen("tcp", ":8080")
	if err != nil {
		log.Fatal(err)
	}

	server := grpc.NewServer()
	proto.RegisterGreeterServer(server, &Server{})
	if err := server.Serve(lis); err != nil {
		log.Fatal(err)
	}
}

服务结构体嵌入 UnimplementedGreeterServer,可以在后续新增 RPC 时保持向前兼容。正式代码还应处理监听和 Serve 返回的错误。

3.4 实现客户端 ​

go
func main() {
	conn, err := grpc.NewClient(
		"localhost:8080",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := proto.NewGreeterClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	response, err := client.SayHello(
		ctx,
		&proto.HelloRequest{Name: "Zhang"},
	)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(response.GetMsg())
}

grpc.NewClient 只创建逻辑上的客户端连接,不会立即发起网络 I/O;通常在第一次 RPC 时才开始解析地址并建立连接。示例使用 insecure.NewCredentials() 便于本机测试,它不会启用 TLS,不能直接照搬到生产环境。

4 数据流模式 ​

gRPC 有四种调用模式。单个流内的消息保持发送顺序,但不同 RPC 之间没有统一的到达顺序。当前版本的 protoc-gen-go-grpc 会为流式方法生成带泛型参数的接口;旧版生成代码中的接口名称可能不同。

4.1 单一请求-响应 ​

客户端发送一个请求,服务端返回一个响应。

proto 定义:

proto
service Greeter{
	rpc GetUser(UserRequest) returns (UserResponse);
}

Go 实现:

go
// 服务端
func (s *Server) GetUser(ctx context.Context, req *pb.UserRequest) (*pb.UserResponse, error) {
    return &pb.UserResponse{...}, nil
}

// 客户端
res, err := client.GetUser(ctx, &pb.UserRequest{...})

4.2 服务端流式响应 ​

客户端发送一个请求,服务端按顺序返回多个响应。

proto 定义:

proto
service Logger{
	rpc StreamLogs(LogRequest) returns (stream LogResponse);
}

Go 实现:

go
// 服务端
func (s *Server) StreamLogs(req *pb.LogRequest, stream grpc.ServerStreamingServer[pb.LogResponse]) error {
    for {
        if err := stream.Send(&pb.LogResponse{...}); err != nil {
            return err
        }
        time.Sleep(1 * time.Second)
    }
}

// 客户端
stream, err := client.StreamLogs(ctx, &pb.LogRequest{...})
for {
    res, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        return err
    }
    log.Println(res)
}

4.3 客户端流式请求 ​

客户端连续发送多个请求,发送结束后由服务端返回一个汇总响应。客户端应调用 CloseAndRecv,服务端则在读到 io.EOF 后调用 SendAndClose。

proto 定义:

proto
service Uploader{
	rpc UploadFile(stream FileChunk) returns (FileSummary);
}

Go 实现:

go
// 服务端
func (s *Server) UploadFile(stream grpc.ClientStreamingServer[pb.FileChunk, pb.FileSummary]) error {
    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.FileSummary{...})
        }
        if err != nil {
            return err
        }
        // 处理 chunk
    }
}

// 客户端
stream, err := client.UploadFile(ctx)
for _, chunk := range chunks {
    if err := stream.Send(chunk); err != nil {
        break
    }
}
summary, err := stream.CloseAndRecv()

4.4 双向流式 ​

客户端和服务端都可以连续发送消息。gRPC-Go 允许一个 goroutine 负责 Send、另一个 goroutine 负责 Recv,但不能让多个 goroutine 同时调用同一方向的方法。

proto 定义:

proto
service Chatter{
	rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

Go 实现:

go
// 服务端
func (s *Server) Chat(stream grpc.BidiStreamingServer[pb.ChatMessage, pb.ChatMessage]) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }
        // 处理消息并回复
        if err := stream.Send(&pb.ChatMessage{...}); err != nil {
            return err
        }
    }
}

// 客户端
stream, err := client.Chat(ctx)
// 发送协程
go func() {
    for {
        stream.Send(&pb.ChatMessage{...})
    }
}()
// 接收协程
go func() {
    for {
        res, err := stream.Recv()
        // 处理响应
    }
}()

4.5 示例 ​

.proto 文件定义:

proto
syntax = "proto3";

package grpcdemo;

option go_package = "example.com/grpcdemo/proto";

service StreamService {
  rpc GetStream(StreamReqData) returns (stream StreamResData);
  rpc PutStream(stream StreamReqData) returns (StreamResData);
  rpc AllStream(stream StreamReqData) returns (stream StreamResData);
}

message StreamReqData {
  string data = 1;
}

message StreamResData {
  string data = 1;
}

服务端:

go
type Server struct {
	proto.UnimplementedStreamServiceServer
}

func (s *Server) GetStream(req *proto.StreamReqData, stream grpc.ServerStreamingServer[proto.StreamResData]) error {
	fmt.Println("====== 服务端 流模式 ======")
	fmt.Println(req.GetData())
	for i := 0; i < 10; i++ {
		if err := stream.Send(&proto.StreamResData{
			Data: "Server: " + time.Now().Format("2006-01-02 15:04:05"),
		}); err != nil {
			return err
		}
		time.Sleep(1 * time.Second)
	}
	return nil
}

func (s *Server) PutStream(stream grpc.ClientStreamingServer[proto.StreamReqData, proto.StreamResData]) error {
	fmt.Println("====== 客户端 流模式 ======")
	count := 0
	for {
		if req, err := stream.Recv(); err == io.EOF {
			return stream.SendAndClose(&proto.StreamResData{
				Data: fmt.Sprintf("received %d messages", count),
			})
		} else if err != nil {
			return err
		} else {
			fmt.Println(req.GetData())
			count++
		}
	}
}

func (s *Server) AllStream(stream grpc.BidiStreamingServer[proto.StreamReqData, proto.StreamResData]) error {
	fmt.Println("======  双向  流模式 ======")
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		fmt.Println(req.GetData())
		if err := stream.Send(&proto.StreamResData{
			Data: "Server: " + time.Now().Format("2006-01-02 15:04:05"),
		}); err != nil {
			return err
		}
	}
}

func main() {
	server := grpc.NewServer()
	proto.RegisterStreamServiceServer(server, &Server{})
	lis, err := net.Listen("tcp", ":8080")
	if err != nil {
		panic(err)
	}
	defer func() {
		if err = lis.Close(); err != nil {
			panic(err)
		}
	}()
	if err = server.Serve(lis); err != nil {
		panic(err)
	}
}

客户端:

go
func main() {
	conn, err := grpc.NewClient(
		"localhost:8080",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := conn.Close(); err != nil {
			panic(err)
		}
	}()

	client := proto.NewStreamServiceClient(conn)
	ctx := context.Background()

	fmt.Println("====== 服务端 流模式 ======")
	stream1, err := client.GetStream(ctx, &proto.StreamReqData{
		Data: "Client: " + time.Now().Format("2006-01-02 15:04:05"),
	})
	if err != nil {
		panic(err)
	}
	for {
		if res, err := stream1.Recv(); err == io.EOF {
			break
		} else if err != nil {
			panic(err)
		} else {
			fmt.Println(res.GetData())
		}
	}

	fmt.Println("====== 客户端 流模式 ======")
	stream2, err := client.PutStream(ctx)
	if err != nil {
		panic(err)
	}
	for i := 0; i < 10; i++ {
		if err = stream2.Send(&proto.StreamReqData{
			Data: "Client: " + time.Now().Format("2006-01-02 15:04:05"),
		}); err != nil {
			panic(err)
		}
		time.Sleep(1 * time.Second)
	}
	res2, err := stream2.CloseAndRecv()
	if err != nil {
		panic(err)
	}
	fmt.Println(res2.GetData())

	fmt.Println("======  双向  流模式 ======")
	stream3, err := client.AllStream(ctx)
	if err != nil {
		panic(err)
	}
	sendErr := make(chan error, 1)

	go func() {
		for i := 0; i < 10; i++ {
			if err := stream3.Send(&proto.StreamReqData{
				Data: "Client: " + time.Now().Format("2006-01-02 15:04:05"),
			}); err != nil {
				sendErr <- err
				return
			}
			time.Sleep(1 * time.Second)
		}
		sendErr <- stream3.CloseSend()
	}()

	for {
		res, err := stream3.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			panic(err)
		}
		fmt.Println(res.GetData())
	}
	if err := <-sendErr; err != nil {
		panic(err)
	}
}

流结束时要区分 io.EOF 与真正的传输错误。服务端处理函数应返回错误,不要在处理请求的 goroutine 中直接 panic,否则一次异常调用可能终止整个进程。

5 元数据 ​

元数据用于传递认证信息、追踪 ID 等调用上下文,在线上对应 HTTP/2 的 header 和 trailer。键不区分大小写,不能以保留前缀 grpc- 开头;普通值应使用 ASCII,二进制键以 -bin 结尾。

5.1 构建元数据 ​

可以使用包 google.golang.org/grpc/metadata 创建元数据。类型 MD 实际上是 map:

go
type MD map[string][]string

元数据可以像 map 一样读取,其中每个键对应一组值。它不适合承载大块业务数据,过大的 header 可能被客户端、代理或服务端拒绝。

5.1.1 创建新元数据 ​

可以使用函数 New 从 map[string]string 创建元数据:

go
md := metadata.New(map[string]string{"key1": "val1", "key2": "val2"})

另一种方法是使用 Pairs。具有相同键的值将被合并到一个切片中:

go
md := metadata.Pairs(
    "key1", "val1",
    "key1", "val1-2", // "key1" 的值将变成 []string{"val1", "val1-2"}
    "key2", "val2",
)

注意

所有的键都会自动转换为小写, 因此 key1 和 kEy1 是相同的键,它们的值将被合并到同一个切片中。无论是 New 还是 Pairs 都是这样的。

5.1.2 在元数据中存储二进制数据 ​

在 Go API 中,元数据的键和值都是字符串。传递原始字节时,把字节转换为字符串,并给键添加 -bin 后缀;gRPC 会处理线上的二进制编码:

go
md := metadata.Pairs(
    "key", "string value",
    "key-bin", string([]byte{96, 102}),  // 二进制数据在发送前将被编码(base64),并在传输后被解码。
)

5.2 客户端发送和接收 ​

5.2.1 发送元数据 ​

有两种方法可以将元数据发送到服务端。推荐的方法是使用 AppendToOutgoingContext 将键值对追加到上下文中。当不存在元数据时,则添加元数据;当上下文中已存在元数据时,将合并键值对。

go
// 创建新上下文,并添加一些元数据
ctx := metadata.AppendToOutgoingContext(ctx, "k1", "v1", "k1", "v2", "k2", "v3")

// 向上下文中添加更多元数据
ctx := metadata.AppendToOutgoingContext(ctx, "k3", "v4")

// 一元RPC
response, err := client.SomeRPC(ctx, someRequest)

// 流式RPC
stream, err := client.SomeStreamingRPC(ctx)

也可以使用 NewOutgoingContext,但是这会替换上下文中的现有元数据。

go
// 创建新上下文,并添加一些元数据
md := metadata.Pairs("k1", "v1", "k1", "v2", "k2", "v3")
ctx := metadata.NewOutgoingContext(context.Background(), md)

// 向上下文中添加更多元数据
send, _ := metadata.FromOutgoingContext(ctx)
newMD := metadata.Pairs("k3", "v3")
ctx = metadata.NewOutgoingContext(ctx, metadata.Join(send, newMD))

// 一元RPC
response, err := client.SomeRPC(ctx, someRequest)

// 流式RPC
stream, err := client.SomeStreamingRPC(ctx)

5.2.2 接收元数据 ​

一元调用:

go
var header, trailer metadata.MD
r, err := client.SomeRPC(
    ctx,
    someRequest,
    grpc.Header(&header),
    grpc.Trailer(&trailer),
)

流式调用中,Header 会等待响应头到达;Trailer 应在读取到 io.EOF 后获取:

go
stream, err := client.SomeStreamingRPC(ctx)
if err != nil {
    return err
}
header, err := stream.Header()
if err != nil {
    return err
}
for {
    _, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        return err
    }
}
trailer := stream.Trailer()

5.3 服务端发送和接收 ​

5.3.1 接收元数据 ​

如果是一元调用,则可以使用 RPC 处理程序的上下文。对于流式调用,服务端需要从流中获取上下文。

一元调用:

go
func (s *server) SomeRPC(ctx context.Context, in *pb.SomeRequest) (*pb.SomeResponse, error) {
    md, ok := metadata.FromIncomingContext(ctx)
}

流式调用:

go
func (s *server) SomeStreamingRPC(stream pb.Service_SomeStreamingRPCServer) error {
    md, ok := metadata.FromIncomingContext(stream.Context())
}

5.3.2 发送元数据 ​

一元调用可以使用 grpc.SetHeader 和 grpc.SetTrailer。它们接收当前 RPC 的上下文,返回的错误不应忽略:

go
func (s *server) SomeRPC(ctx context.Context, in *pb.SomeRequest) (*pb.SomeResponse, error) {
	header := metadata.Pairs("header-key", "val")
	if err := grpc.SetHeader(ctx, header); err != nil {
		return nil, err
	}
	trailer := metadata.Pairs("trailer-key", "val")
	if err := grpc.SetTrailer(ctx, trailer); err != nil {
		return nil, err
	}
	return &pb.SomeResponse{}, nil
}

对于流式调用,可以使用接口 ServerStream 中的函数 SetHeader 和 SetTrailer 发送 header 和 trailer:

go
func (s *server) SomeStreamingRPC(stream pb.Service_SomeStreamingRPCServer) error {
    header := metadata.Pairs("header-key", "val")
    if err := stream.SetHeader(header); err != nil {
        return err
    }
    trailer := metadata.Pairs("trailer-key", "val")
    stream.SetTrailer(trailer)
    return nil
}

重要

服务端不能通过 metadata.NewOutgoingContext 发送响应元数据,应使用 grpc.SetHeader、grpc.SetTrailer 或流对象上的对应方法。

6 拦截器 ​

拦截器位于应用代码与 gRPC 调用之间,常用于日志、认证、指标和追踪。客户端与服务端分别提供一元和流式拦截器,共四种类型。拦截器必须继续调用传入的 invoker、streamer 或 handler,除非它明确要提前拒绝这次调用。

6.1 客户端拦截器 ​

6.1.1 一元拦截器 ​

客户端一元拦截器的类型为 UnaryClientInterceptor。它本质上是一个带有签名的函数类型:

go
type UnaryClientInterceptor func(ctx context.Context, method string, req, reply any, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error

一元拦截器通常分为三步:预处理、调用 RPC 方法和后处理。

对于预处理,可以通过检查传入的参数来获取有关当前 RPC 调用的信息。参数包括 RPC 上下文、方法字符串、要发送的请求和配置的 CallOptions。有了这些信息甚至可以修改 RPC 调用。

预处理后,用户可以通过调用 invoker 来调用 RPC 调用。

一旦调用程序返回,就可以对 RPC 调用进行后处理。这通常涉及处理返回的回复和错误。

要在 ClientConn 上安装一元拦截器,可以给 NewClient 传入 WithUnaryInterceptor。需要组合多个拦截器时使用 WithChainUnaryInterceptor,执行顺序与注册顺序一致。

6.1.2 流拦截器 ​

客户端流拦截器的类型是 StreamClientInterceptor。它是一个带有签名的函数类型:

go
type StreamClientInterceptor func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error)

流拦截器的实现通常包括预处理和流操作拦截。

预处理类似于一元拦截器。

然而,流拦截器不是调用 RPC 方法然后进行后处理,而是拦截对流的操作。拦截器首先调用传入的 streamer 以获取 ClientStream,然后在用拦截逻辑重载其方法的同时封装 ClientStream。最后,拦截器将包装好的 ClientStream 返回。

流拦截器通过 WithStreamInterceptor 安装,多个流拦截器可以使用 WithChainStreamInterceptor 组合。

6.1.3 示例 ​

go
func interceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	// 拦截下来统计耗时
	start := time.Now()
	err := invoker(ctx, method, req, reply, cc, opts...)  // invoker 就是原先的调用逻辑
	fmt.Printf("耗时:%s", time.Since(start))
	return err
}

func main() {
	conn, err := grpc.NewClient(
		"localhost:8080",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithUnaryInterceptor(interceptor),
	)
}

6.2 服务端拦截器 ​

服务端拦截器与客户端拦截器相似,但提供的参数信息略有不同。

6.2.1 一元拦截器 ​

服务端一元拦截器的类型是 UnaryServerInterceptor。它是一个带有签名的函数类型:

go
type UnaryServerInterceptor func(ctx context.Context, req any, info *UnaryServerInfo, handler UnaryHandler) (resp any, err error)

服务端一元拦截器通过 UnaryInterceptor 安装,多个拦截器使用 ChainUnaryInterceptor。

6.2.2 流拦截器 ​

服务端流拦截器的类型是 StreamServerInterceptor。它是一个具有签名的函数类型:

go
type StreamServerInterceptor func(srv any, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error

服务端流拦截器通过 StreamInterceptor 安装,多个拦截器使用 ChainStreamInterceptor。

6.2.3 示例 ​

go
func interceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
	// 拦截下来统计耗时
	start := time.Now()
	resp, err = handler(ctx, req)  // handler 就是原来正常的处理逻辑
	fmt.Printf("耗时:%s", time.Since(start))
	return resp, err
}

func main() {
	opt := grpc.UnaryInterceptor(interceptor)
	server := grpc.NewServer(opt)
}

7 错误处理 ​

gRPC 错误由状态码、说明文本和可选详情组成。服务端应返回稳定、可判断的状态码,不要把数据库错误或内部堆栈直接暴露给客户端。客户端先判断状态码,再决定提示用户、重试还是终止调用。

7.1 状态码 ​

状态码ID描述
OK0成功时返回
CANCELLED1操作被取消,通常是由调用者取消的
UNKNOWN2未知错误
INVALID_ARGUMENT3客户端指定的参数无效
DEADLINE_EXCEEDED4操作超时
NOT_FOUND5找不到某些请求的实体(例如文件或目录)
ALREADY_EXISTS6客户端尝试创建的实体(例如,文件或目录)已存在
PERMISSION_DENIED7调用者没有执行指定操作的权限
RESOURCE_EXHAUSTED8某些资源已耗尽,可能是用户的配额或者文件系统空间不足
FAILED_PRECONDITION9操作被拒绝,因为系统未处于执行操作所需的状态
ABORTED10操作被中止
OUT_OF_RANGE11操作已超出有效范围
UNIMPLEMENTED12此服务中未实现或不支持/未启用该操作
INTERNAL13内部错误
UNAVAILABLE14该服务当前不可用
DATA_LOSS15无法恢复的数据丢失或损坏
UNAUTHENTICATED16请求没有操作的有效身份验证凭据

7.2 服务端 ​

go
func (s *Server) SayHello(ctx context.Context, in *proto.HelloRequest) (*proto.HelloResponse, error) {
	err := status.New(codes.NotFound, "Not Found").Err()
	return nil, err
}

7.3 客户端 ​

go
if _, err := client.SayHello(context.Background(), nil); err != nil {
	if st, ok := status.FromError(err); ok {
		fmt.Println(st.Code(), st.Message())
	}
}

8 超时与取消 ​

gRPC 默认不会自动给调用设置 deadline。客户端应根据业务耗时设置超时,并在结束后调用 cancel 释放计时器:

go
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
_, err := client.SayHello(ctx, &proto.HelloRequest{})
if err != nil {
	if st, ok := status.FromError(err); ok {
		fmt.Println(st.Code())     // DeadlineExceeded
		fmt.Println(st.Message())  // context deadline exceeded
	}
}

客户端取消上下文或超过 deadline 后,服务端上下文也会被取消。耗时操作应定期检查 ctx.Err() 或 ctx.Done(),否则业务 goroutine 可能在调用已经失败后继续占用资源。

9 上线前检查 ​

  • 传输安全:本机示例可以使用 insecure.NewCredentials(),跨主机部署应配置 TLS,并校验服务端身份。
  • 优雅退出:停止接收新请求后调用 GracefulStop,给正在执行的 RPC 留出收尾时间;超时后再决定是否强制 Stop。
  • 健康检查:使用 gRPC 标准健康检查服务,不要用一个业务 RPC 代替实例健康状态。
  • 反射服务:grpcurl 等工具依赖服务反射。生产环境是否开启,应根据接口暴露范围决定。
  • 请求边界:为消息大小、并发数和执行时间设置合理限制,不能只依赖框架默认值。

Released under the MIT License.