grpc-development

gRPC 开发

本技能介绍使用 Protocol Buffers 构建基于 gRPC 的服务和 API 的最佳实践，包括服务设计、流模式、拦截器、安全性和可观测性。

构建 gRPC 服务的工作流

1. 定义服务契约 — 编写 .proto 文件，其中包含服务定义、RPC 方法和消息类型，并遵循下文的样式与命名约定。
2. 生成语言存根 — 使用相应的语言插件（例如 protoc-gen-go-grpc、grpcio-tools）运行 protoc，生成服务端和客户端代码。
3. 实现服务端 — 为每个 RPC 方法创建处理函数，将其注册到 gRPC 服务端，并配置 TLS、拦截器和健康检查。
4. 实现客户端 — 创建到服务端的通道，实例化生成的客户端存根，并在设置适当的截止时间和错误处理后调用 RPC 方法。
5. 添加拦截器 — 接入服务端和客户端拦截器，以实现日志记录、身份验证、指标收集和跟踪。
6. 编写测试 — 使用模拟依赖对处理函数进行单元测试；通过真实的 gRPC 连接进行集成测试。
7. 部署并观测 — 在投入生产之前，启用分布式跟踪（OpenTelemetry）、结构化日志记录和指标仪表板。

核心原则

• gRPC 使用 Protocol Buffers 作为接口定义语言（IDL）和消息交换格式
• 围绕定义可远程调用的方法及其参数和返回类型这一理念来设计服务
• 优先考虑类型安全、性能和向后兼容性
• 实现中不得留下待办事项、占位符或缺失部分

Protocol Buffer 最佳实践

文件组织（1-1-1 模式）

• 每个 .proto 文件中只定义一个顶层实体（message、enum 或 extension）
• 每个 .proto 文件对应一个构建规则
• 这样可以使 proto 定义保持精简、模块化
• 优点包括简化重构、缩短构建时间和减小二进制文件体积

消息设计

• 使用结构化消息以便扩展——Protocol Buffers 支持添加字段，且不会破坏现有客户端
• 对于以后可能需要添加字段的结构，应谨慎使用 struct
• 不要在多个 RPC 之间重复使用消息——API 可能会随时间变化，应避免让不同的 RPC 调用紧密耦合
• 字段之间应始终彼此独立——不要让一个字段影响另一个字段的语义

字段指南

• 使用描述性字段名，并采用 underscore_separated_names 格式
• 为已删除的字段保留字段编号，以防止未来发生冲突
• 对于可能并非总是存在的字段，使用 optional
• 当用户需要在互斥选项之间进行选择时，可考虑使用 oneof

Enum 最佳实践

• 确保第一个值始终为 0
• 使用 "UNSPECIFIED" 作为默认值（例如，STATUS_UNSPECIFIED = 0）
• 使用前缀避免命名冲突（例如，ORDER_STATUS_CREATED 与 STATUS_PENDING）
• 为已移除的 enum 值保留编号，以防意外重复使用

样式指南

• 行长度保持在 80 个字符以内
• 字符串优先使用双引号
• 包名应使用小写字母
• 消息名使用 CamelCase（首字母大写）
• 字段名使用 underscore_separated_names 格式
• 服务名和 RPC 方法名使用 CamelCase

服务设计

RPC 模式

• 一元 RPC：客户端发送单个请求，服务端返回单个响应
• 服务端流式 RPC：客户端发送请求，服务端以消息流的形式响应
• 客户端流式 RPC：客户端发送消息流，服务端返回单个响应
• 双向流式 RPC：双方都发送消息流

示例：Proto 定义

syntax = "proto3";

package order.v1;

option go_package = "gen/order/v1;orderv1";

// OrderService manages customer orders.
service OrderService {
  // Creates a new order and returns the created resource.
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);
  // Streams real-time status updates for an order.
  rpc WatchOrder(WatchOrderRequest) returns (stream OrderStatus);
}

message CreateOrderRequest {
  string customer_id = 1;
  repeated OrderItem items = 2;
}

message CreateOrderResponse {
  string order_id = 1;
  OrderStatus status = 2;
}

message WatchOrderRequest {
  string order_id = 1;
}

message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
}

message OrderStatus {
  string order_id = 1;
  OrderState state = 2;
  string updated_at = 3;
}

enum OrderState {
  ORDER_STATE_UNSPECIFIED = 0;
  ORDER_STATE_CREATED = 1;
  ORDER_STATE_PROCESSING = 2;
  ORDER_STATE_SHIPPED = 3;
  ORDER_STATE_DELIVERED = 4;
}

示例：Go 服务端实现

package main

import (
	"context"
	"log"
	"net"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	pb "example.com/gen/order/v1"
)

type orderServer struct {
	pb.UnimplementedOrderServiceServer
}

func (s *orderServer) CreateOrder(ctx context.Context, req *pb.CreateOrderRequest) (*pb.CreateOrderResponse, error) {
	if req.GetCustomerId() == "" {
		return nil, status.Error(codes.InvalidArgument, "customer_id is required")
	}
	orderID := "ord-" + time.Now().Format("20060102150405")
	return &pb.CreateOrderResponse{
		OrderId: orderID,
		Status: &pb.OrderStatus{
			OrderId: orderID,
			State:   pb.OrderState_ORDER_STATE_CREATED,
		},
	}, nil
}

func (s *orderServer) WatchOrder(req *pb.WatchOrderRequest, stream pb.OrderService_WatchOrderServer) error {
	for i, state := range []pb.OrderState{
		pb.OrderState_ORDER_STATE_PROCESSING,
		pb.OrderState_ORDER_STATE_SHIPPED,
		pb.OrderState_ORDER_STATE_DELIVERED,
	} {
		select {
		case <-stream.Context().Done():
			return stream.Context().Err()
		case <-time.After(time.Duration(i) * time.Second):
			if err := stream.Send(&pb.OrderStatus{
				OrderId:   req.GetOrderId(),
				State:     state,
				UpdatedAt: time.Now().Format(time.RFC3339),
			}); err != nil {
				return err
			}
		}
	}
	return nil
}

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	srv := grpc.NewServer(
		grpc.UnaryInterceptor(loggingUnaryInterceptor),
	)
	pb.RegisterOrderServiceServer(srv, &orderServer{})
	log.Println("serving on :50051")
	if err := srv.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

// loggingUnaryInterceptor logs each unary RPC call.
func loggingUnaryInterceptor(
	ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (any, error) {
	start := time.Now()
	resp, err := handler(ctx, req)
	log.Printf("method=%s duration=%s err=%v", info.FullMethod, time.Since(start), err)
	return resp, err
}

API 设计

• 设计清晰、直观的服务接口
• 将相关方法归入同一服务
• 使用有意义的方法名来描述操作
• 为每个 RPC 编写注释，说明其行为、参数和返回值

性能优化

通道管理

• 使用 gRPC 时复用通道
• 创建 gRPC 通道的成本较高，因为这会新建一个 HTTP/2 连接
• 在高吞吐场景中实现连接池
• 适当配置 keepalive 设置

消息优化

• 将消息大小控制在合理范围内——大型消息会影响性能
• 对于大型数据传输，可考虑使用流式传输
• 在带宽受限的环境中使用压缩
• 避免消息结构嵌套过深

错误处理

状态码

• 使用适当的 gRPC 状态码（OK、INVALID_ARGUMENT、NOT_FOUND 等）
• 在状态详情中包含有意义的错误消息
• 对于复杂的错误场景，使用丰富的错误详情
• 在服务定义中记录预期的错误情况

重试逻辑

• 对瞬时故障使用指数退避重试
• 为所有 RPC 调用设置截止时间或超时
• 对 UNAVAILABLE 和 RESOURCE_EXHAUSTED 错误进行重试
• 不要盲目重试非幂等操作

安全

身份验证

• 在生产环境中使用 TLS 保障传输安全
• 使用元数据／标头实现逐 RPC 身份验证
• 支持多种身份验证机制（JWT、OAuth2、mTLS）
• 对每个请求验证凭据

授权

• 实现方法级访问控制
• 使用拦截器集中处理授权逻辑
• 无论身份验证状态如何，都要验证所有输入数据
• 遵循最小权限原则

拦截器和中间件

服务端拦截器

• 使用拦截器处理横切关注点（日志记录、身份验证、指标）
• 谨慎排列拦截器顺序——执行顺序很重要
• 让拦截器专注于单一职责
• 在拦截器中妥善处理错误

客户端拦截器

• 添加用于追踪和身份验证的元数据（标头）
• 实现请求／响应日志记录
• 添加自动重试逻辑
• 收集客户端指标

测试

单元测试

• 模拟 gRPC 服务，以便进行隔离测试
• 测试消息的序列化／反序列化
• 验证错误处理路径
• 独立测试拦截器逻辑

集成测试

• 尽可能使用真实的 gRPC 连接进行测试
• 端到端验证流式传输行为
• 测试超时和取消场景
• 使用符合实际的流量模式进行负载测试

可观测性

分布式追踪

• 使用 OpenTelemetry 跨服务边界进行分布式追踪
• 在元数据中传递追踪上下文
• 为客户端和服务端都添加埋点
• 为每个 RPC 调用创建 Span

指标

• 跟踪 RPC 延迟直方图
• 按方法和状态码监控错误率
• 统计活跃连接数和流数量
• 针对异常和 SLA 违规情况设置告警

日志记录

• 使用字段一致的结构化日志
• 记录 RPC 方法、耗时和状态
• 包含追踪 ID 以便关联
• 避免记录敏感数据

特定语言指南

Go

• 使用官方的 google.golang.org/grpc 包
• 将服务实现为接口类型
• 使用 context 处理取消和截止时间
• 使用 protoc-gen-go-grpc 进行代码生成

Python

• 使用 grpcio 和 grpcio-tools 包
• 使用 grpcio-aio 实现异步服务，以提高并发能力
• 将类型提示与生成的存根结合使用
• 在异步上下文中妥善处理阻塞调用

Node.js/TypeScript

• 使用 @grpc/grpc-js（纯 JavaScript 实现）
• 可考虑使用 nice-grpc，以获得更好的 TypeScript 支持
• 利用 async/await 模式
• 使用静态代码生成确保类型安全