grpc-microservices

gRPC 微服务

这是一套全面的技能，用于使用 gRPC 和 Protocol Buffers 构建高性能、类型安全的微服务。本技能涵盖服务设计、所有流模式、拦截器、负载均衡、错误处理，以及分布式系统的生产部署模式。

何时使用本技能

在以下情况下使用本技能：

• 构建需要高性能、低延迟通信的微服务
• 实现服务间的实时数据流
• 使用 Protocol Buffers 设计具有强契约的类型安全 API
• 创建由不同语言编写服务的多语言系统
• 构建需要双向流的分布式系统
• 使用高级路由和可观测性实现服务网格
• 设计需要向后／向前兼容演进的 API
• 创建对性能和类型安全要求较高的内部 API
• 使用流式数据管道构建事件驱动架构
• 实现具备推送能力的客户端－服务器系统（服务端流）
• 设计需要高效二进制序列化的系统
• 构建需要为多种语言自动生成代码的微服务

核心概念

gRPC 基础知识

gRPC 是一种现代开源 RPC 框架，可在任何环境中运行。它使客户端和服务器应用能够透明地通信，并让构建互联系统变得更加容易。

主要特性：

• 基于 HTTP/2：多路复用、服务器推送、标头压缩
• Protocol Buffers：高效的二进制序列化格式
• 流式传输：内置双向流支持
• 代码生成：在 10 多种语言中自动生成客户端／服务器代码
• 截止时间／超时：一流的超时支持
• 取消：在服务间传播取消信号
• 拦截器：用于横切关注点的中间件模式

Protocol Buffers（protobuf）

Protocol Buffers 是一种与语言和平台无关的可扩展机制，用于序列化结构化数据。

优势：

• 紧凑：比 JSON 小 3～10 倍
• 快速：序列化／反序列化速度比 JSON 快 20～100 倍
• 类型安全：具备验证功能的强类型架构
• 向后／向前兼容：安全地演进架构
• 语言支持：官方支持 10 多种语言
• 自描述：架构本身可作为文档

基本语法：

syntax = "proto3";

message User {
  int32 id = 1;
  string name = 2;
  string email = 3;
}

服务定义

gRPC 服务在 .proto 文件中定义，并指定可用方法及其输入／输出类型。

基本服务：

service UserService {
  rpc GetUser(GetUserRequest) returns (GetUserResponse);
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);
}

四种 RPC 方法

1. 一元 RPC（请求－响应）

简单的请求－响应模式，类似传统的 REST API 调用。

rpc GetUser(GetUserRequest) returns (GetUserResponse);

适用场景：

• CRUD 操作
• 简单查询
• 同步操作
• 传统请求－响应模式

2. 服务端流式 RPC

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

rpc ListUsers(ListUsersRequest) returns (stream User);

适用场景：

• 分页结果
• 实时更新
• 服务端事件推送
• 大型数据集下载

3. 客户端流式 RPC

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

rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

适用场景：

• 批量上传
• 批处理
• 客户端聚合
• 分块上传文件

4. 双向流式 RPC

客户端和服务端分别独立发送一系列消息。

rpc Chat(stream ChatMessage) returns (stream ChatMessage);

适用场景：

• 实时聊天应用
• 实时协作
• 游戏（实时状态同步）
• IoT 双向通信

Protobuf 架构设计

消息设计最佳实践

1. 使用明确的字段编号

字段编号对向后兼容至关重要，绝不能重复使用。

message User {
  int32 id = 1;           // Never change this number
  string name = 2;        // Never change this number
  string email = 3;       // Never change this number
  // int32 age = 4;       // DEPRECATED - don't reuse 4
  string phone = 5;       // New field - use next available
}

2. 使用枚举表示固定集合

enum UserRole {
  USER_ROLE_UNSPECIFIED = 0;  // Always have a zero value
  USER_ROLE_ADMIN = 1;
  USER_ROLE_MODERATOR = 2;
  USER_ROLE_MEMBER = 3;
}

message User {
  int32 id = 1;
  string name = 2;
  UserRole role = 3;
}

3. 使用嵌套消息表示复杂类型

message User {
  int32 id = 1;
  string name = 2;

  message Address {
    string street = 1;
    string city = 2;
    string state = 3;
    string zip = 4;
  }

  Address address = 3;
  repeated Address additional_addresses = 4;
}

4. 使用 repeated 表示数组

message UserList {
  repeated User users = 1;
}

message User {
  int32 id = 1;
  string name = 2;
  repeated string tags = 3;
}

5. 使用 oneof 表示联合类型

message SearchRequest {
  string query = 1;

  oneof filter {
    string category = 2;
    int32 user_id = 3;
    string tag = 4;
  }
}

6. 使用 google.protobuf 内置常用类型

import "google/protobuf/timestamp.proto";
import "google/protobuf/duration.proto";
import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";

message Event {
  string id = 1;
  string name = 2;
  google.protobuf.Timestamp created_at = 3;
  google.protobuf.Duration duration = 4;
  google.protobuf.Int32Value optional_count = 5;
}

服务设计模式

1. 面向资源的设计

遵循适用于 RPC 的 RESTful 原则：

service UserService {
  // Get single resource
  rpc GetUser(GetUserRequest) returns (User);

  // List resources
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);

  // Create resource
  rpc CreateUser(CreateUserRequest) returns (User);

  // Update resource
  rpc UpdateUser(UpdateUserRequest) returns (User);

  // Delete resource
  rpc DeleteUser(DeleteUserRequest) returns (google.protobuf.Empty);
}

2. 分页模式

message ListUsersRequest {
  int32 page_size = 1;
  string page_token = 2;
  string filter = 3;
}

message ListUsersResponse {
  repeated User users = 1;
  string next_page_token = 2;
  int32 total_count = 3;
}

3. 批量操作模式

message BatchGetUsersRequest {
  repeated int32 user_ids = 1;
}

message BatchGetUsersResponse {
  map<int32, User> users = 1;
  repeated int32 not_found = 2;
}

4. 长时间运行操作模式

import "google/longrunning/operations.proto";

service BatchJobService {
  rpc ProcessBatch(BatchRequest) returns (google.longrunning.Operation);
  rpc GetOperation(GetOperationRequest) returns (google.longrunning.Operation);
}

流式传输模式

服务端流式传输模式

1. 分页流式传输

高效地流式传输大型结果集：

service ProductService {
  rpc SearchProducts(SearchRequest) returns (stream Product);
}

message SearchRequest {
  string query = 1;
  int32 limit = 2;
}

实现（Go）：

func (s *server) SearchProducts(req *pb.SearchRequest, stream pb.ProductService_SearchProductsServer) error {
    products := s.db.Search(req.Query, req.Limit)

    for _, product := range products {
        if err := stream.Send(&product); err != nil {
            return err
        }
    }

    return nil
}

2. 实时更新

在更新发生时将其推送给客户端：

service EventService {
  rpc SubscribeToEvents(SubscribeRequest) returns (stream Event);
}

message SubscribeRequest {
  repeated string event_types = 1;
  google.protobuf.Timestamp since = 2;
}

3. 日志跟踪

流式传输日志或审计跟踪：

service LogService {
  rpc TailLogs(TailRequest) returns (stream LogEntry);
}

message TailRequest {
  string service_name = 1;
  string level = 2;
  int32 lines = 3;
}

客户端流式传输模式

1. 批量上传

客户端流式传输数据，服务端处理并返回摘要：

service UploadService {
  rpc UploadImages(stream ImageChunk) returns (UploadSummary);
}

message ImageChunk {
  string filename = 1;
  bytes data = 2;
  int32 chunk_number = 3;
}

message UploadSummary {
  int32 total_images = 1;
  int64 total_bytes = 2;
  repeated string uploaded_filenames = 3;
}

实现（Go）：

func (s *server) UploadImages(stream pb.UploadService_UploadImagesServer) error {
    var count int32
    var totalBytes int64
    var filenames []string

    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.UploadSummary{
                TotalImages: count,
                TotalBytes:  totalBytes,
                UploadedFilenames: filenames,
            })
        }
        if err != nil {
            return err
        }

        // Process chunk
        totalBytes += int64(len(chunk.Data))
        if chunk.ChunkNumber == 0 {
            count++
            filenames = append(filenames, chunk.Filename)
        }
    }
}

2. 聚合

客户端发送多个数据点，服务端进行聚合：

service AnalyticsService {
  rpc RecordMetrics(stream Metric) returns (AggregateResult);
}

message Metric {
  string name = 1;
  double value = 2;
  google.protobuf.Timestamp timestamp = 3;
}

双向流式传输模式

1. 聊天应用

实时双向通信：

service ChatService {
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message ChatMessage {
  string user_id = 1;
  string room_id = 2;
  string content = 3;
  google.protobuf.Timestamp timestamp = 4;
}

实现（Go）：

func (s *server) Chat(stream pb.ChatService_ChatServer) error {
    // Create channel for this client
    clientID := uuid.New().String()
    msgChan := make(chan *pb.ChatMessage, 10)

    // Register client
    s.mu.Lock()
    s.clients[clientID] = msgChan
    s.mu.Unlock()

    defer func() {
        s.mu.Lock()
        delete(s.clients, clientID)
        close(msgChan)
        s.mu.Unlock()
    }()

    // Goroutine to send messages to client
    go func() {
        for msg := range msgChan {
            if err := stream.Send(msg); err != nil {
                return
            }
        }
    }()

    // Receive messages from client
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        // Broadcast to all clients in room
        s.broadcast(msg)
    }
}

2. 实时协作

实时编辑文档：

service CollaborationService {
  rpc Collaborate(stream DocumentEdit) returns (stream DocumentEdit);
}

message DocumentEdit {
  string document_id = 1;
  string user_id = 2;
  int32 position = 3;
  string content = 4;
  enum Operation {
    OPERATION_UNSPECIFIED = 0;
    OPERATION_INSERT = 1;
    OPERATION_DELETE = 2;
    OPERATION_UPDATE = 3;
  }
  Operation operation = 5;
}

3. 游戏状态同步

实时多人游戏更新：

service GameService {
  rpc PlayGame(stream GameAction) returns (stream GameState);
}

message GameAction {
  string player_id = 1;
  string game_id = 2;
  string action_type = 3;
  bytes action_data = 4;
}

message GameState {
  string game_id = 1;
  repeated PlayerState players = 2;
  bytes world_state = 3;
  google.protobuf.Timestamp timestamp = 4;
}

拦截器（中间件）

拦截器提供了一种为 gRPC 服务添加横切关注点的方法。

一元拦截器

服务端一元拦截器：

func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        // Pre-processing
        start := time.Now()
        log.Printf("Method: %s, Start: %v", info.FullMethod, start)

        // Call the handler
        resp, err := handler(ctx, req)

        // Post-processing
        duration := time.Since(start)
        log.Printf("Method: %s, Duration: %v, Error: %v",
            info.FullMethod, duration, err)

        return resp, err
    }
}

// Usage
server := grpc.NewServer(
    grpc.UnaryInterceptor(UnaryServerInterceptor()),
)

客户端一元拦截器：

func UnaryClientInterceptor() grpc.UnaryClientInterceptor {
    return func(
        ctx context.Context,
        method string,
        req, reply interface{},
        cc *grpc.ClientConn,
        invoker grpc.UnaryInvoker,
        opts ...grpc.CallOption,
    ) error {
        start := time.Now()

        // Call the remote method
        err := invoker(ctx, method, req, reply, cc, opts...)

        log.Printf("Method: %s, Duration: %v, Error: %v",
            method, time.Since(start), err)

        return err
    }
}

// Usage
conn, err := grpc.Dial(
    address,
    grpc.WithUnaryInterceptor(UnaryClientInterceptor()),
)

流式拦截器

服务端流拦截器：

func StreamServerInterceptor() grpc.StreamServerInterceptor {
    return func(
        srv interface{},
        ss grpc.ServerStream,
        info *grpc.StreamServerInfo,
        handler grpc.StreamHandler,
    ) error {
        log.Printf("Stream started: %s", info.FullMethod)

        err := handler(srv, ss)

        log.Printf("Stream ended: %s, Error: %v", info.FullMethod, err)

        return err
    }
}

常见拦截器模式

1. 身份验证拦截器

func AuthInterceptor(secret string) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        // Extract metadata
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, status.Error(codes.Unauthenticated, "no metadata")
        }

        // Get authorization token
        tokens := md["authorization"]
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "no token")
        }

        // Validate token
        token := tokens[0]
        claims, err := validateJWT(token, secret)
        if err != nil {
            return nil, status.Error(codes.Unauthenticated, "invalid token")
        }

        // Add claims to context
        ctx = context.WithValue(ctx, "claims", claims)

        return handler(ctx, req)
    }
}

2. 日志记录拦截器

func LoggingInterceptor(logger *log.Logger) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        start := time.Now()

        // Get request ID from metadata
        requestID := getRequestID(ctx)

        logger.Printf("[%s] Request: %s", requestID, info.FullMethod)

        resp, err := handler(ctx, req)

        duration := time.Since(start)
        statusCode := status.Code(err)

        logger.Printf("[%s] Response: %s, Duration: %v, Status: %v",
            requestID, info.FullMethod, duration, statusCode)

        return resp, err
    }
}

3. 速率限制拦截器

func RateLimitInterceptor(limiter *rate.Limiter) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        if !limiter.Allow() {
            return nil, status.Error(
                codes.ResourceExhausted,
                "rate limit exceeded",
            )
        }

        return handler(ctx, req)
    }
}

4. 跟踪拦截器（OpenTelemetry）

func TracingInterceptor(tracer trace.Tracer) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        ctx, span := tracer.Start(ctx, info.FullMethod)
        defer span.End()

        // Add attributes
        span.SetAttributes(
            attribute.String("rpc.method", info.FullMethod),
            attribute.String("rpc.service", "MyService"),
        )

        resp, err := handler(ctx, req)

        if err != nil {
            span.RecordError(err)
            span.SetStatus(codes2.Error, err.Error())
        } else {
            span.SetStatus(codes2.Ok, "")
        }

        return resp, err
    }
}

5. 错误恢复拦截器

func RecoveryInterceptor() grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (resp interface{}, err error) {
        defer func() {
            if r := recover(); r != nil {
                log.Printf("Panic recovered: %v\n%s", r, debug.Stack())
                err = status.Error(codes.Internal, "internal server error")
            }
        }()

        return handler(ctx, req)
    }
}

链接多个拦截器

server := grpc.NewServer(
    grpc.ChainUnaryInterceptor(
        RecoveryInterceptor(),
        LoggingInterceptor(logger),
        TracingInterceptor(tracer),
        AuthInterceptor(jwtSecret),
        RateLimitInterceptor(limiter),
    ),
    grpc.ChainStreamInterceptor(
        StreamRecoveryInterceptor(),
        StreamLoggingInterceptor(logger),
    ),
)

负载均衡

客户端负载均衡

gRPC 内置了多种策略的客户端负载均衡功能。

1. 轮询

import "google.golang.org/grpc/balancer/roundrobin"

conn, err := grpc.Dial(
    "dns:///my-service.example.com:50051",
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
    grpc.WithInsecure(),
)

2. 优先选择第一个（默认）

conn, err := grpc.Dial(
    "dns:///my-service.example.com:50051",
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"pick_first"}`),
    grpc.WithInsecure(),
)

3. 自定义解析器

实现自定义服务发现：

type exampleResolver struct {
    target     resolver.Target
    cc         resolver.ClientConn
    addrsStore map[string][]string
}

func (r *exampleResolver) ResolveNow(resolver.ResolveNowOptions) {
    // Discover service addresses
    addresses := r.discoverServices()

    var addrs []resolver.Address
    for _, addr := range addresses {
        addrs = append(addrs, resolver.Address{Addr: addr})
    }

    r.cc.UpdateState(resolver.State{Addresses: addrs})
}

func init() {
    resolver.Register(&exampleResolverBuilder{})
}

使用服务网格进行负载均衡

使用服务网格的 Kubernetes（Istio/Linkerd）：

apiVersion: v1
kind: Service
metadata:
  name: grpc-service
spec:
  selector:
    app: grpc-app
  ports:
  - name: grpc
    port: 50051
    targetPort: 50051
    protocol: TCP
  type: ClusterIP
---
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: grpc-service
spec:
  host: grpc-service
  trafficPolicy:
    loadBalancer:
      simple: ROUND_ROBIN
    connectionPool:
      http:
        http2MaxRequests: 1000
        maxRequestsPerConnection: 10

健康检查

实现健康检查服务：

service Health {
  rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
  rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
}

message HealthCheckRequest {
  string service = 1;
}

message HealthCheckResponse {
  enum ServingStatus {
    UNKNOWN = 0;
    SERVING = 1;
    NOT_SERVING = 2;
    SERVICE_UNKNOWN = 3;
  }
  ServingStatus status = 1;
}

实现：

import "google.golang.org/grpc/health"
import healthpb "google.golang.org/grpc/health/grpc_health_v1"

healthServer := health.NewServer()
healthpb.RegisterHealthServer(grpcServer, healthServer)

// Set service status
healthServer.SetServingStatus("UserService", healthpb.HealthCheckResponse_SERVING)

错误处理

gRPC 状态码

gRPC 使用标准化状态码处理错误：

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

// Return errors with appropriate codes
func (s *server) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
    if req.Id <= 0 {
        return nil, status.Error(codes.InvalidArgument, "id must be positive")
    }

    user, err := s.db.GetUser(req.Id)
    if err == sql.ErrNoRows {
        return nil, status.Error(codes.NotFound, "user not found")
    }
    if err != nil {
        return nil, status.Error(codes.Internal, "database error")
    }

    return user, nil
}

常见状态码：

• OK：成功
• Canceled：操作已取消
• Unknown：未知错误
• InvalidArgument：客户端指定了无效参数
• DeadlineExceeded：操作完成前已超出截止时间
• NotFound：未找到实体
• AlreadyExists：实体已存在
• PermissionDenied：权限被拒绝
• ResourceExhausted：资源耗尽（速率限制）
• FailedPrecondition：操作被拒绝（系统状态无效）
• Aborted：操作已中止
• OutOfRange：超出有效范围
• Unimplemented：操作尚未实现
• Internal：内部服务器错误
• Unavailable：服务不可用
• DataLoss：发生不可恢复的数据丢失
• Unauthenticated：请求缺少有效的身份验证信息

丰富的错误详情

添加结构化的错误详情：

import "google.golang.org/genproto/googleapis/rpc/errdetails"

func (s *server) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {
    // Validate request
    violations := validateCreateUserRequest(req)
    if len(violations) > 0 {
        badRequest := &errdetails.BadRequest{}
        for field, msg := range violations {
            badRequest.FieldViolations = append(
                badRequest.FieldViolations,
                &errdetails.BadRequest_FieldViolation{
                    Field:       field,
                    Description: msg,
                },
            )
        }

        st := status.New(codes.InvalidArgument, "invalid request")
        st, _ = st.WithDetails(badRequest)
        return nil, st.Err()
    }

    // Create user...
}

客户端错误处理：

resp, err := client.CreateUser(ctx, req)
if err != nil {
    st := status.Convert(err)

    for _, detail := range st.Details() {
        switch t := detail.(type) {
        case *errdetails.BadRequest:
            for _, violation := range t.FieldViolations {
                fmt.Printf("Invalid field %s: %s\n",
                    violation.Field, violation.Description)
            }
        }
    }
}

错误传播

func (s *server) ProcessOrder(ctx context.Context, req *pb.OrderRequest) (*pb.OrderResponse, error) {
    // Call inventory service
    inventory, err := s.inventoryClient.CheckInventory(ctx, &pb.InventoryRequest{
        ProductId: req.ProductId,
    })
    if err != nil {
        // Propagate error with additional context
        st := status.Convert(err)
        return nil, status.Errorf(st.Code(),
            "inventory check failed: %v", st.Message())
    }

    // Continue processing...
}

重试逻辑

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

func CallWithRetry(ctx context.Context, maxRetries int, fn func() error) error {
    var err error

    for i := 0; i < maxRetries; i++ {
        err = fn()
        if err == nil {
            return nil
        }

        // Check if error is retryable
        st := status.Convert(err)
        if !isRetryable(st.Code()) {
            return err
        }

        // Exponential backoff
        backoff := time.Duration(math.Pow(2, float64(i))) * time.Second
        time.Sleep(backoff)
    }

    return err
}

func isRetryable(code codes.Code) bool {
    return code == codes.Unavailable ||
           code == codes.DeadlineExceeded ||
           code == codes.ResourceExhausted
}

最佳实践

1. 模式演进

应该：

• 始终使用 syntax = "proto3"
• 切勿重复使用字段编号
• 对弃用的字段使用 reserved
• 为新增字段分配新的编号
• 对可空字段使用可选包装类型

message User {
  int32 id = 1;
  string name = 2;
  // string age = 3;  // Deprecated
  reserved 3;
  reserved "age";

  string email = 4;
  google.protobuf.Int32Value phone = 5;  // Optional
}

不应该：

• 更改字段类型
• 重复使用字段编号
• 移除字段却不保留其编号
• 更改消息名称却不添加别名

2. 性能优化

连接管理：

// Reuse connections
var conn *grpc.ClientConn
var once sync.Once

func getConnection() *grpc.ClientConn {
    once.Do(func() {
        var err error
        conn, err = grpc.Dial(
            address,
            grpc.WithKeepaliveParams(keepalive.ClientParameters{
                Time:                10 * time.Second,
                Timeout:             3 * time.Second,
                PermitWithoutStream: true,
            }),
        )
        if err != nil {
            log.Fatal(err)
        }
    })
    return conn
}

连接池：

type ConnectionPool struct {
    connections []*grpc.ClientConn
    next        uint32
}

func (p *ConnectionPool) GetConnection() *grpc.ClientConn {
    n := atomic.AddUint32(&p.next, 1)
    return p.connections[n%uint32(len(p.connections))]
}

对大型数据使用流式传输：

// Instead of this:
rpc GetAllUsers(Empty) returns (UserList);  // Large response

// Use this:
rpc ListUsers(ListUsersRequest) returns (stream User);  // Streamed

3. 安全最佳实践

TLS 配置：

// Server-side
creds, err := credentials.NewServerTLSFromFile(certFile, keyFile)
server := grpc.NewServer(grpc.Creds(creds))

// Client-side
creds, err := credentials.NewClientTLSFromFile(certFile, "")
conn, err := grpc.Dial(address, grpc.WithTransportCredentials(creds))

双向 TLS（mTLS）：

cert, err := tls.LoadX509KeyPair(certFile, keyFile)
certPool := x509.NewCertPool()
ca, err := ioutil.ReadFile(caFile)
certPool.AppendCertsFromPEM(ca)

creds := credentials.NewTLS(&tls.Config{
    Certificates: []tls.Certificate{cert},
    ClientAuth:   tls.RequireAndVerifyClientCert,
    ClientCAs:    certPool,
})

server := grpc.NewServer(grpc.Creds(creds))

令牌身份验证：

type tokenAuth struct {
    token string
}

func (t tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
    return map[string]string{
        "authorization": "Bearer " + t.token,
    }, nil
}

func (t tokenAuth) RequireTransportSecurity() bool {
    return true
}

// Usage
conn, err := grpc.Dial(
    address,
    grpc.WithPerRPCCredentials(tokenAuth{token: "my-token"}),
)

4. 超时与截止时间管理

// Set deadline for request
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

resp, err := client.GetUser(ctx, &pb.GetUserRequest{Id: 123})
if err != nil {
    if status.Code(err) == codes.DeadlineExceeded {
        log.Println("Request timed out")
    }
}

服务端截止时间传播：

func (s *server) ComplexOperation(ctx context.Context, req *pb.Request) (*pb.Response, error) {
    // Check if deadline is already exceeded
    deadline, ok := ctx.Deadline()
    if ok && time.Now().After(deadline) {
        return nil, status.Error(codes.DeadlineExceeded, "deadline exceeded")
    }

    // Propagate context to downstream calls
    user, err := s.userClient.GetUser(ctx, &pb.GetUserRequest{Id: req.UserId})
    if err != nil {
        return nil, err
    }

    // Continue with remaining time...
}

5. 监控与可观测性

Prometheus 指标：

import "github.com/grpc-ecosystem/go-grpc-prometheus"

// Server metrics
grpcMetrics := grpc_prometheus.NewServerMetrics()
server := grpc.NewServer(
    grpc.UnaryInterceptor(grpcMetrics.UnaryServerInterceptor()),
    grpc.StreamInterceptor(grpcMetrics.StreamServerInterceptor()),
)
grpcMetrics.InitializeMetrics(server)

// Expose metrics
http.Handle("/metrics", promhttp.Handler())

6. 优雅关闭

server := grpc.NewServer()
// Register services...

go func() {
    if err := server.Serve(listener); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}()

// Wait for interrupt signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, os.Interrupt, syscall.SIGTERM)
<-quit

log.Println("Shutting down server...")

// Graceful shutdown
server.GracefulStop()
log.Println("Server stopped")

7. 服务版本控制

基于 URL 的版本控制：

package api.v1;

service UserServiceV1 {
  rpc GetUser(GetUserRequest) returns (User);
}

package api.v2;

service UserServiceV2 {
  rpc GetUser(GetUserRequest) returns (User);
}

基于字段的版本控制：

message User {
  int32 id = 1;
  string name = 2;
  string email = 3;

  // v2 additions
  string phone = 4;
  Address address = 5;
}

8. 测试最佳实践

使用 Mock 进行单元测试：

type mockUserClient struct {
    pb.UserServiceClient
    getUserFunc func(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error)
}

func (m *mockUserClient) GetUser(ctx context.Context, req *pb.GetUserRequest, opts ...grpc.CallOption) (*pb.User, error) {
    return m.getUserFunc(ctx, req)
}

func TestOrderService(t *testing.T) {
    mockClient := &mockUserClient{
        getUserFunc: func(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
            return &pb.User{Id: 1, Name: "Test User"}, nil
        },
    }

    // Test with mock...
}

集成测试：

func TestIntegration(t *testing.T) {
    // Start test server
    lis, err := net.Listen("tcp", ":0")
    require.NoError(t, err)

    server := grpc.NewServer()
    pb.RegisterUserServiceServer(server, &userServer{})

    go server.Serve(lis)
    defer server.Stop()

    // Connect client
    conn, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure())
    require.NoError(t, err)
    defer conn.Close()

    client := pb.NewUserServiceClient(conn)

    // Test requests...
}

生产环境部署模式

Docker 部署

Dockerfile：

FROM golang:1.21-alpine AS builder

WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o server ./cmd/server

FROM alpine:latest
RUN apk --no-cache add ca-certificates
WORKDIR /root/

COPY --from=builder /app/server .
COPY --from=builder /app/proto ./proto

EXPOSE 50051

CMD ["./server"]

Kubernetes 部署

deployment.yaml：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: grpc-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: grpc-service
  template:
    metadata:
      labels:
        app: grpc-service
    spec:
      containers:
      - name: grpc-service
        image: grpc-service:latest
        ports:
        - containerPort: 50051
          name: grpc
          protocol: TCP
        env:
        - name: PORT
          value: "50051"
        livenessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:50051"]
          initialDelaySeconds: 10
        readinessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:50051"]
          initialDelaySeconds: 5
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
---
apiVersion: v1
kind: Service
metadata:
  name: grpc-service
spec:
  selector:
    app: grpc-service
  ports:
  - port: 50051
    targetPort: 50051
    protocol: TCP
    name: grpc
  type: ClusterIP

服务网格集成（Istio）

用于流量路由的 VirtualService：

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: grpc-service
spec:
  hosts:
  - grpc-service
  http:
  - match:
    - headers:
        version:
          exact: v2
    route:
    - destination:
        host: grpc-service
        subset: v2
  - route:
    - destination:
        host: grpc-service
        subset: v1

常见模式与反模式

✅ 应该做：

1. 使用流式传输处理大型数据集
2. 使用状态码实现适当的错误处理
3. 为横切关注点添加拦截器
4. 为高吞吐量客户端使用连接池
5. 实现健康检查
6. 设置适当的超时和截止时间
7. 在生产环境中使用 TLS
8. 为 API 设置版本
9. 通过指标和追踪进行监控
10. 使用集成测试进行测试

❌ 不要做：

1. 不要使用一元 RPC 处理大型数据集 - 改用流式传输
2. 不要忽略上下文取消 - 始终检查 context.Done()
3. 不要为每个请求创建新连接 - 复用连接
4. 不要跳过身份验证/授权 - 始终进行验证
5. 不要忘记优雅关闭 - 正确处理 SIGTERM
6. 不要硬编码端点 - 使用服务发现
7. 不要忽略错误 - 处理所有错误情况
8. 不要在没有超时的情况下执行阻塞操作 - 始终设置截止时间
9. 不要跳过健康检查 - 实现存活/就绪探针
10. 不要在没有监控的情况下部署 - 添加指标和日志记录

---

技能版本：1.0.0 最后更新：2025 年 10 月 技能类别：微服务、gRPC、分布式系统、API 设计 兼容：Go、Python、Node.js、Java、C++、C#、Ruby 等