为什么选gRPC

两年前做一个实时性要求高的项目，最开始用的RESTful API。随着调用量上来，HTTP/1.1的开销越来越明显：

• JSON序列化/反序列化吃CPU
• 每个请求都要建立连接（短连接）
• Header开销大
• P99延迟抖动厉害

后来换成gRPC：

• Protocol Buffers二进制序列化，小而快
• HTTP/2多路复用，一个连接跑多个请求
• 支持流式通信
• 强类型，接口定义清晰

切换后，P99从80ms降到15ms，CPU使用率降了30%。代价是调试没那么方便了（二进制协议不如JSON直观）。

基础概念

gRPC有四种通信模式：

1. Unary：一问一答，最常用
2. Server Streaming：客户端一个请求，服务端返回流
3. Client Streaming：客户端发送流，服务端一个响应
4. Bidirectional Streaming：双向流

对于大多数业务场景，Unary就够用了。流式用在实时数据推送、大文件传输等场景。

项目结构

先看个实际的项目结构：

.
├── proto/                    # 协议定义
│   ├── user/
│   │   └── user.proto
│   └── order/
│       └── order.proto
├── gen/                      # 生成的代码
│   ├── go/
│   │   ├── user/
│   │   └── order/
│   └── java/
├── services/
│   ├── user-service/
│   └── order-service/
├── buf.yaml                  # buf配置
└── buf.gen.yaml

Proto设计规范

好的Proto定义能省很多事。

基本规范

// user/user.proto
syntax = "proto3";

package user.v1;

option go_package = "github.com/example/gen/go/user/v1;userv1";
option java_package = "com.example.user.v1";
option java_multiple_files = true;

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

// 用户服务
service UserService {
  // 获取用户信息
  rpc GetUser(GetUserRequest) returns (GetUserResponse);
  
  // 批量获取用户
  rpc BatchGetUsers(BatchGetUsersRequest) returns (BatchGetUsersResponse);
  
  // 更新用户
  rpc UpdateUser(UpdateUserRequest) returns (UpdateUserResponse);
  
  // 用户列表（分页）
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
}

// 用户信息
message User {
  int64 id = 1;
  string username = 2;
  string email = 3;
  string phone = 4;
  UserStatus status = 5;
  google.protobuf.Timestamp created_at = 6;
  google.protobuf.Timestamp updated_at = 7;
}

enum UserStatus {
  USER_STATUS_UNSPECIFIED = 0;
  USER_STATUS_ACTIVE = 1;
  USER_STATUS_INACTIVE = 2;
  USER_STATUS_BANNED = 3;
}

message GetUserRequest {
  int64 user_id = 1;
}

message GetUserResponse {
  User user = 1;
}

message BatchGetUsersRequest {
  repeated int64 user_ids = 1;
}

message BatchGetUsersResponse {
  repeated User users = 1;
}

message UpdateUserRequest {
  User user = 1;
  // 只更新指定字段
  google.protobuf.FieldMask update_mask = 2;
}

message UpdateUserResponse {
  User user = 1;
}

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

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

几个关键点

1. 包名带版本：package user.v1，方便后续兼容
2. 枚举第一个是UNSPECIFIED：这是最佳实践，避免0值歧义
3. 用FieldMask做部分更新：不用每次传完整对象
4. 分页用page_token：比offset性能好

用buf管理Proto

buf比protoc好用，依赖管理、lint、breaking change检测都有：

# buf.yaml
version: v1
breaking:
  use:
    - FILE
lint:
  use:
    - DEFAULT
  except:
    - PACKAGE_VERSION_SUFFIX
deps:
  - buf.build/googleapis/googleapis

# buf.gen.yaml
version: v1
managed:
  enabled: true
  go_package_prefix:
    default: github.com/example/gen/go
plugins:
  - plugin: buf.build/protocolbuffers/go
    out: gen/go
    opt: paths=source_relative
  - plugin: buf.build/grpc/go
    out: gen/go
    opt: paths=source_relative

生成代码：

buf generate

Go服务端实现

package main

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

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "google.golang.org/grpc/keepalive"
    
    userv1 "github.com/example/gen/go/user/v1"
)

type userServer struct {
    userv1.UnimplementedUserServiceServer
    repo UserRepository
}

func (s *userServer) GetUser(ctx context.Context, req *userv1.GetUserRequest) (*userv1.GetUserResponse, error) {
    if req.UserId <= 0 {
        return nil, status.Error(codes.InvalidArgument, "user_id is required")
    }

    user, err := s.repo.GetByID(ctx, req.UserId)
    if err != nil {
        if errors.Is(err, ErrNotFound) {
            return nil, status.Error(codes.NotFound, "user not found")
        }
        return nil, status.Error(codes.Internal, "internal error")
    }

    return &userv1.GetUserResponse{
        User: toProtoUser(user),
    }, nil
}

func (s *userServer) BatchGetUsers(ctx context.Context, req *userv1.BatchGetUsersRequest) (*userv1.BatchGetUsersResponse, error) {
    if len(req.UserIds) == 0 {
        return nil, status.Error(codes.InvalidArgument, "user_ids is required")
    }
    if len(req.UserIds) > 100 {
        return nil, status.Error(codes.InvalidArgument, "too many user_ids, max 100")
    }

    users, err := s.repo.BatchGetByIDs(ctx, req.UserIds)
    if err != nil {
        return nil, status.Error(codes.Internal, "internal error")
    }

    protoUsers := make([]*userv1.User, 0, len(users))
    for _, u := range users {
        protoUsers = append(protoUsers, toProtoUser(u))
    }

    return &userv1.BatchGetUsersResponse{
        Users: protoUsers,
    }, nil
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    // 服务端配置
    opts := []grpc.ServerOption{
        grpc.KeepaliveParams(keepalive.ServerParameters{
            MaxConnectionIdle:     15 * time.Minute,
            MaxConnectionAge:      30 * time.Minute,
            MaxConnectionAgeGrace: 5 * time.Minute,
            Time:                  5 * time.Minute,
            Timeout:               1 * time.Minute,
        }),
        grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
            MinTime:             5 * time.Minute,
            PermitWithoutStream: true,
        }),
        grpc.MaxRecvMsgSize(4 * 1024 * 1024),  // 4MB
        grpc.MaxSendMsgSize(4 * 1024 * 1024),
    }

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

    log.Println("gRPC server listening on :50051")
    if err := server.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

Go客户端实现

package main

import (
    "context"
    "log"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/keepalive"
    
    userv1 "github.com/example/gen/go/user/v1"
)

func main() {
    // 客户端配置
    opts := []grpc.DialOption{
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithKeepaliveParams(keepalive.ClientParameters{
            Time:                10 * time.Second,
            Timeout:             3 * time.Second,
            PermitWithoutStream: true,
        }),
        grpc.WithDefaultCallOptions(
            grpc.MaxCallRecvMsgSize(4*1024*1024),
            grpc.MaxCallSendMsgSize(4*1024*1024),
        ),
    }

    conn, err := grpc.Dial("localhost:50051", opts...)
    if err != nil {
        log.Fatalf("failed to dial: %v", err)
    }
    defer conn.Close()

    client := userv1.NewUserServiceClient(conn)

    // 调用示例
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    resp, err := client.GetUser(ctx, &userv1.GetUserRequest{
        UserId: 12345,
    })
    if err != nil {
        log.Fatalf("GetUser failed: %v", err)
    }

    log.Printf("User: %v", resp.User)
}

拦截器实现

拦截器是gRPC的中间件机制，用来做日志、监控、认证等。

服务端拦截器

// 日志拦截器
func LoggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    start := time.Now()
    
    resp, err := handler(ctx, req)
    
    duration := time.Since(start)
    log.Printf("method=%s duration=%v err=%v", info.FullMethod, duration, err)
    
    return resp, err
}

// 恢复拦截器（防止panic导致服务挂掉）
func RecoveryInterceptor(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 error")
        }
    }()
    return handler(ctx, req)
}

// 认证拦截器
func AuthInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return nil, status.Error(codes.Unauthenticated, "missing metadata")
    }

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

    userID, err := validateToken(tokens[0])
    if err != nil {
        return nil, status.Error(codes.Unauthenticated, "invalid token")
    }

    // 把用户ID放到context里
    ctx = context.WithValue(ctx, "user_id", userID)
    return handler(ctx, req)
}

// 使用
server := grpc.NewServer(
    grpc.ChainUnaryInterceptor(
        RecoveryInterceptor,
        LoggingInterceptor,
        AuthInterceptor,
    ),
)

客户端拦截器

// 超时拦截器
func TimeoutInterceptor(timeout time.Duration) grpc.UnaryClientInterceptor {
    return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
        ctx, cancel := context.WithTimeout(ctx, timeout)
        defer cancel()
        return invoker(ctx, method, req, reply, cc, opts...)
    }
}

// 重试拦截器
func RetryInterceptor(maxRetries int) grpc.UnaryClientInterceptor {
    return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
        var lastErr error
        for i := 0; i < maxRetries; i++ {
            err := invoker(ctx, method, req, reply, cc, opts...)
            if err == nil {
                return nil
            }
            
            // 只重试可重试的错误
            if !isRetryable(err) {
                return err
            }
            
            lastErr = err
            time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
        }
        return lastErr
    }
}

func isRetryable(err error) bool {
    s, ok := status.FromError(err)
    if !ok {
        return false
    }
    switch s.Code() {
    case codes.Unavailable, codes.ResourceExhausted, codes.Aborted:
        return true
    default:
        return false
    }
}

流式通信

实时数据推送场景用Server Streaming：

service NotificationService {
  // 订阅通知
  rpc Subscribe(SubscribeRequest) returns (stream Notification);
}

服务端：

func (s *notificationServer) Subscribe(req *notificationv1.SubscribeRequest, stream notificationv1.NotificationService_SubscribeServer) error {
    userID := req.UserId
    
    // 创建通道
    ch := make(chan *notificationv1.Notification, 100)
    s.subscribers.Store(userID, ch)
    defer s.subscribers.Delete(userID)

    for {
        select {
        case <-stream.Context().Done():
            return nil
        case notification := <-ch:
            if err := stream.Send(notification); err != nil {
                return err
            }
        }
    }
}

客户端：

stream, err := client.Subscribe(ctx, &notificationv1.SubscribeRequest{
    UserId: 12345,
})
if err != nil {
    log.Fatal(err)
}

for {
    notification, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Received: %v", notification)
}

生产环境配置

健康检查

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

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

// 设置服务健康状态
healthServer.SetServingStatus("user.v1.UserService", healthpb.HealthCheckResponse_SERVING)

优雅关闭

func main() {
    server := grpc.NewServer()
    
    // 注册服务...
    
    // 优雅关闭
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    
    go func() {
        <-quit
        log.Println("shutting down server...")
        
        // 停止接收新连接，等待现有请求完成
        server.GracefulStop()
    }()
    
    if err := server.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

负载均衡

gRPC默认用pick_first，生产环境要改成round_robin：

conn, err := grpc.Dial(
    "dns:///user-service.default.svc.cluster.local:50051",
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
)

或者用服务发现：

import "google.golang.org/grpc/resolver"

// 注册自定义resolver
resolver.Register(&consulResolver{})

conn, err := grpc.Dial("consul://user-service")

踩过的坑

1. 连接池问题

症状：偶发的UNAVAILABLE错误。

原因：gRPC客户端默认只建立一个HTTP/2连接，连接断开时会出错。

解决：配置重连和keepalive

grpc.WithKeepaliveParams(keepalive.ClientParameters{
    Time:                10 * time.Second,
    Timeout:             3 * time.Second,
    PermitWithoutStream: true,
})

2. Context超时传递

症状：调用链路上游超时了，下游还在傻傻执行。

原因：Context没有正确传递。

解决：永远用请求的Context，不要用context.Background()

3. 大消息传输失败

症状：传大对象报错"grpc: received message larger than max"。

原因：默认消息大小限制是4MB。

解决：调大限制，或者改用流式传输

grpc.WithDefaultCallOptions(
    grpc.MaxCallRecvMsgSize(16*1024*1024),  // 16MB
)

4. 序列化耗时

症状：CPU使用率高，主要消耗在protobuf序列化。

原因：消息太大或结构太复杂。

解决：

• 减少不必要的字段
• 用bytes字段传原始数据（跳过序列化）
• 批量接口减少调用次数

跨网络调用

gRPC基于HTTP/2，默认需要长连接。如果服务部署在不同的网络环境（比如多个机房），网络延迟和稳定性会影响性能。

遇到过一个场景：两个机房之间的gRPC调用延迟很高。后来用星空组网打通网络后，延迟稳定在可接受范围内。对于需要跨网络调用的场景，保证底层网络质量很重要。

总结

gRPC适合的场景：

• 内部微服务通信
• 对延迟敏感的场景
• 流式数据传输
• 多语言环境

不太适合的场景：

• 对外开放的API（还是REST/OpenAPI方便）
• 浏览器直接调用（需要grpc-web代理）
• 简单的CRUD（杀鸡焉用牛刀）

实践建议：

1. Proto定义要规范，前期投入后期省心
2. 拦截器做好日志、监控、认证
3. 生产环境配置keepalive和超时
4. 错误处理用status包，别用普通error
5. 版本管理用package版本号，别改proto字段号