gRPC折腾手记

作者: 醉月思发布日期: 2021-06-05 分类: 技术实践

先说清楚 gRPC 在协议层面的真实面貌：它是 HTTP/2 之上的 RPC 框架，使用 protobuf 作为接口定义语言（IDL）和序列化格式。

但"HTTP/2"这个前提容易被忽略。

为什么从 REST 换到 gRPC

最先触发我认真考虑 gRPC 的是流式传输。

那时候在做一个实时数据处理服务，需要从多个数据源持续接收更新。用 REST 的轮询或者 WebSocket 都不是最佳解：轮询延迟高、连接开销大；WebSocket 能双向通信但协议层面没有结构定义，全靠约定。

gRPC 的双向流正好卡在这个场景上。而且 protobuf 的二进制序列化确实比 JSON 紧凑不少，尤其是传输大量数值数据时。

另一个触发点是跨语言调用。Go 后端、Python 数据处理、Java 业务逻辑，三种语言的接口用 OpenAPI 定义倒也能跑，但每次改接口都要同步更新三方文档和代码生成，很容易出遗漏。protobuf 的 .proto 文件成了单一定义源，各语言代码生成统一管理，这块省了不少事。

当然，换技术栈从来不是纯粹技术问题。当时的团队已经习惯了 REST 的调试工具、链路追踪、监控指标，这些在 gRPC 体系下都变了。折腾下来，最大的感受是：技术升级容易，团队对齐难。

gRPC 协议其实没那么复杂

先说清楚 gRPC 在协议层面的真实面貌：它是 HTTP/2 之上的 RPC 框架，使用 protobuf 作为接口定义语言（IDL）和序列化格式。

但"HTTP/2"这个前提容易被忽略。很多人以为 gRPC 就是"二进制的 REST"，其实它依赖 HTTP/2 的多路复用、头部压缩、服务器推送等特性。这就是为什么 gRPC 不会直接跑在 HTTP/1.1 上。

protobuf 的序列化性能确实不错，但不是所有场景都快。小消息时解析开销占比高，大消息时 protobuf 的紧凑格式才有明显优势。我们测试过：100 字节以内的消息，JSON 和 protobuf 的序列化时间差距在微秒级，但网络传输差异几乎可以忽略。

syntax = "proto3";

package example;

service DataStream {
  // 单次请求-响应
  rpc GetData (GetDataRequest) returns (GetDataResponse);

  // 服务端流式
  rpc StreamData (StreamRequest) returns (stream DataChunk);

  // 客户端流式
  rpc UploadData (stream DataChunk) returns (UploadResponse);

  // 双向流式
  rpc BidirectionalStream (stream StreamRequest) returns (stream DataChunk);
}

message GetDataRequest {
  string data_id = 1;
  int64 version = 2;
}

message GetDataResponse {
  DataChunk data = 1;
  int64 timestamp = 2;
}

message StreamRequest {
  string filter = 1;
  int32 batch_size = 2;
}

message DataChunk {
  bytes payload = 1;
  int32 chunk_id = 2;
}

message UploadResponse {
  bool success = 1;
  string message = 2;
  int64 bytes_received = 3;
}

这个 .proto 文件定义了四种调用模式。实际用起来，单向请求响应最简单，但真正体现 gRPC 优势的是流式调用。数据管道、实时通知、增量同步，这些场景用流式比轮询或长轮询舒服太多。

// server.go
package main

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

    "google.golang.org/grpc"
    pb "path/to/your/proto"
)

type server struct {
    pb.UnimplementedDataStreamServer
}

func (s *server) GetData(ctx context.Context, req *pb.GetDataRequest) (*pb.GetDataResponse, error) {
    // 模拟数据查询
    data := &pb.DataChunk{
        Payload: []byte("sample data"),
        ChunkId: 1,
    }
    return &pb.GetDataResponse{
        Data:      data,
        Timestamp: time.Now().Unix(),
    }, nil
}

func (s *server) StreamData(req *pb.StreamRequest, stream pb.DataStream_StreamDataServer) error {
    for i := 0; i < 10; i++ {
        chunk := &pb.DataChunk{
            Payload: []byte("chunk data"),
            ChunkId: int32(i),
        }

        if err := stream.Send(chunk); err != nil {
            log.Printf("Send failed: %v", err)
            return err
        }

        time.Sleep(100 * time.Millisecond)
    }
    return nil
}

func (s *server) UploadData(stream pb.DataStream_UploadDataServer) error {
    var totalBytes int64
    var chunkCount int

    for {
        chunk, err := stream.Recv()
        if err != nil {
            break
        }

        totalBytes += int64(len(chunk.Payload))
        chunkCount++
    }

    return stream.SendAndClose(&pb.UploadResponse{
        Success:      true,
        Message:      "Upload completed",
        BytesReceived: totalBytes,
    })
}

func (s *server) BidirectionalStream(stream pb.DataStream_BidirectionalStreamServer) error {
    for {
        req, err := stream.Recv()
        if err != nil {
            break
        }

        // 简单的 echo 逻辑
        response := &pb.DataChunk{
            Payload: []byte("echo: " + req.Filter),
            ChunkId: int32(time.Now().Unix()),
        }

        if err := stream.Send(response); err != nil {
            log.Printf("Send failed: %v", err)
            return err
        }
    }
    return nil
}

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

    s := grpc.NewServer()
    pb.RegisterDataStreamServer(s, &server{})

    log.Println("Server starting on :50051")
    if err := s.Serve(lis); err != nil {
        log.Fatalf("Failed to serve: %v", err)
    }
}

服务端实现没什么花哨的，主要在流式处理的逻辑里。需要注意的点是：流式调用里错误处理要特别小心，一旦出错整个流可能中断，客户端和服务器都要有重连机制。

// client.go
package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    pb "path/to/your/proto"
)

func main() {
    conn, err := grpc.NewClient("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewDataStreamClient(conn)

    // 测试单向调用
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()

    resp, err := client.GetData(ctx, &pb.GetDataRequest{
        DataId: "test-123",
        Version: 1,
    })
    if err != nil {
        log.Fatalf("GetData failed: %v", err)
    }
    fmt.Printf("Received data: %v\n", resp)

    // 测试服务端流
    stream, err := client.StreamData(ctx, &pb.StreamRequest{
        Filter: "sample",
        BatchSize: 10,
    })
    if err != nil {
        log.Fatalf("StreamData failed: %v", err)
    }

    for {
        chunk, err := stream.Recv()
        if err != nil {
            break
        }
        fmt.Printf("Received chunk: %d\n", chunk.ChunkId)
    }

    // 测试客户端流
    uploadStream, err := client.UploadData(ctx)
    if err != nil {
        log.Fatalf("UploadData failed: %v", err)
    }

    for i := 0; i < 5; i++ {
        if err := uploadStream.Send(&pb.DataChunk{
            Payload: []byte(fmt.Sprintf("chunk-%d", i)),
            ChunkId: int32(i),
        }); err != nil {
            log.Fatalf("Send failed: %v", err)
        }
    }

    uploadResp, err := uploadStream.CloseAndRecv()
    if err != nil {
        log.Fatalf("CloseAndRecv failed: %v", err)
    }
    fmt.Printf("Upload response: %v\n", uploadResp)
}

客户端代码也是直截了当。真正复杂的地方不在调用本身，而在连接管理、超时控制、重试策略这些外围机制上。

实际踩过的几个坑

HTTP/2 依赖不是虚的

刚把 gRPC 跑起来时，一切都挺顺利。直到部署到生产环境的 Nginx 后面，客户端一直报错：“http2: server sent GOAWAY and closed the connection”。

查了一圈才发现，Nginx 的默认配置对 HTTP/2 支持有限，尤其是反向代理 gRPC 时需要额外配置。加上这个配置后问题解决：

server {
    listen 443 ssl http2;
    server_name your-domain.com;

    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;

    location / {
        grpc_pass grpc://backend:50051;

        # 关键配置
        grpc_set_header Host $host;
        grpc_set_header X-Real-IP $remote_addr;
        grpc_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        grpc_set_header X-Forwarded-Proto $scheme;

        # 超时设置
        grpc_read_timeout 300s;
        grpc_send_timeout 300s;
    }
}

教训很直接：不要假设网络基础设施天然支持 HTTP/2，尤其经过代理、防火墙、CDN 时。每次变更都要验证端到端的 HTTP/2 连通性。

消息大小限制吃瘪

有次客户端上传大文件，一直报错：“received message larger than max (4194304)"。

这是 gRPC 默认的消息大小限制（4MB）。改配置倒是简单：

opts := []grpc.ServerOption{
    grpc.MaxRecvMsgSize(16 * 1024 * 1024),  // 16MB
    grpc.MaxSendMsgSize(16 * 1024 * 1024),
}

s := grpc.NewServer(opts...)

客户端也得对应调整：

opts := []grpc.DialOption{
    grpc.WithDefaultCallOptions(
        grpc.MaxCallRecvMsgSize(16 * 1024 * 1024),
        grpc.MaxCallSendMsgSize(16 * 1024 * 1024),
    ),
}

conn, err := grpc.NewClient("localhost:50051", opts...)

但更好的做法是避免大消息。分块传输、流式上传、改用共享存储，都比直接扩大消息限制来得优雅。我们后来改成了客户端流式上传，服务端边收边处理，内存占用平滑很多。

超时控制没对齐

最坑的一次是因为超时设置不一致导致的生产事故。

服务端设置的是全局超时 30 秒，但某些查询确实需要更长。客户端没意识到这个限制，一直在调用，但服务端提前超时返回。更糟糕的是，错误信息不够明确，客户端把超时错误当成了业务错误处理，导致数据不一致。

教训是：超时策略必须在服务契约里定义清楚，客户端和服务端要统一理解。最好在 .proto 文件的注释里说明各接口的预期耗时范围：

// GetData: 正常响应时间 < 1s，复杂查询可能到 10s
rpc GetData (GetDataRequest) returns (GetDataResponse);

// ProcessLargeData: 预期耗时 10s-60s，请设置合理超时
rpc ProcessLargeData (LargeDataRequest) returns (ProcessResponse);

客户端根据这些注释设置合理的超时，服务端在实现里也要监控耗时并报警。

序列化兼容性差点翻车

某次上线后，Python 客户端调用失败，Go 客户端正常。追查到 protobuf 版本不一致：服务端用 3.20，客户端用 3.15，某些字段的序列化行为变了。

其实 protobuf 有向后兼容性，但前提是遵守字段编号规则：不要复用已删除的字段编号，新增字段要选足够大的编号。我们之前清理无用字段时，删除了一些编号，后来又复用这些编号加了新字段，正好踩到了兼容性雷区。

修复方式很简单：永远不要复用字段编号，新字段从最大的可用编号开始。但如果服务已经部署了多个版本，升级时要特别注意灰度策略。

错误处理怎么做才对

gRPC 的错误处理跟 REST 的 HTTP 状态码机制不太一样，它有自己的一套状态码。

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

func (s *server) GetData(ctx context.Context, req *pb.GetDataRequest) (*pb.GetDataResponse, error) {
    if req.DataId == "" {
        return nil, status.Error(codes.InvalidArgument, "data_id cannot be empty")
    }

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

    return &pb.GetDataResponse{
        Data: data,
        Timestamp: time.Now().Unix(),
    }, nil
}

常用状态码有：

codes.InvalidArgument：参数错误
codes.NotFound：资源不存在
codes.AlreadyExists：资源已存在
codes.PermissionDenied：权限不足
codes.Unauthenticated：未认证
codes.ResourceExhausted：资源耗尽（如限流）
codes.FailedPrecondition：前置条件不满足
codes.Aborted：操作冲突
codes.OutOfRange：参数超出范围
codes.Unimplemented：功能未实现
codes.Internal：内部错误
codes.Unavailable：服务不可用

客户端解析错误：

resp, err := client.GetData(ctx, req)
if err != nil {
    st, ok := status.FromError(err)
    if !ok {
        log.Printf("Unknown error: %v", err)
        return
    }

    switch st.Code() {
    case codes.NotFound:
        // 处理资源不存在
    case codes.InvalidArgument:
        // 处理参数错误
    case codes.Internal:
        // 处理内部错误，可能需要重试或报警
    default:
        log.Printf("Unexpected error: %s, %s", st.Code(), st.Message())
    }
    return
}

更高级的做法是自定义错误详情，这样可以在错误信息里附带结构化数据：

import (
    "google.golang.org/genproto/googleapis/rpc/errdetails"
    "google.golang.org/grpc/status"
)

func (s *server) GetData(ctx context.Context, req *pb.GetDataRequest) (*pb.GetDataResponse, error) {
    if req.DataId == "" {
        st := status.New(codes.InvalidArgument, "data_id cannot be empty")

        // 添加详细错误信息
        details, err := st.WithDetails(
            &errdetails.BadRequest{
                FieldViolations: []*errdetails.BadRequest_FieldViolation{
                    {
                        Field:       "data_id",
                        Description: "cannot be empty",
                    },
                },
            },
        )
        if err != nil {
            return nil, st.Err()
        }

        return nil, details.Err()
    }

    // ...
}

客户端可以解析这些详情并给出更友好的错误提示。

监控和调试怎么办

gRPC 服务上线后，监控和调试也是重要环节。

Prometheus 指标采集需要 grpc-go 的拦截器：

import (
    "github.com/prometheus/client_golang/prometheus"
    "google.golang.org/grpc"
)

var (
    requestsTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "grpc_requests_total",
            Help: "Total number of RPC requests.",
        },
        []string{"method", "status"},
    )

    requestDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "grpc_request_duration_seconds",
            Help:    "RPC request duration in seconds.",
            Buckets: prometheus.DefBuckets,
        },
        []string{"method"},
    )
)

func init() {
    prometheus.MustRegister(requestsTotal)
    prometheus.MustRegister(requestDuration)
}

func metricsInterceptor(
    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).Seconds()
    method := info.FullMethod

    var status string
    if err != nil {
        st, ok := status.FromError(err)
        if ok {
            status = st.Code().String()
        } else {
            status = "unknown"
        }
        requestsTotal.WithLabelValues(method, status).Inc()
    } else {
        status = "OK"
        requestsTotal.WithLabelValues(method, status).Inc()
    }

    requestDuration.WithLabelValues(method).Observe(duration)

    return resp, err
}

// 使用拦截器
opts := []grpc.ServerOption{
    grpc.ChainUnaryInterceptor(
        loggingInterceptor,
        metricsInterceptor,
        recoveryInterceptor,
    ),
}

s := grpc.NewServer(opts...)

日志拦截器：

func loggingInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    start := time.Now()

    log.Printf("gRPC call: %s started", info.FullMethod)

    resp, err := handler(ctx, req)

    duration := time.Since(start)
    log.Printf("gRPC call: %s completed in %v, error: %v",
        info.FullMethod, duration, err)

    return resp, err
}

恢复拦截器防止 panic 导致整个服务崩溃：

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

func recoveryInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("Recovered from panic: %v", r)
        }
    }()

    return handler(ctx, req)
}

调试工具推荐 grpcurl，它是 gRPC 版的 curl：

# 列出服务
grpcurl -plaintext localhost:50051 list

# 列出方法
grpcurl -plaintext localhost:50051 list example.DataStream

# 描述方法
grpcurl -plaintext localhost:50051 describe example.DataStream.GetData

# 调用方法
grpcurl -plaintext -d '{"data_id":"test-123","version":1}' \
    localhost:50051 example.DataStream.GetData

# 监控流式调用
grpcurl -plaintext -d '{"filter":"sample","batch_size":10}' \
    localhost:50051 example.DataStream/StreamData

这趟折腾到底值不值

回看这两年从 REST 迁移到 gRPC 的过程，只能说"一半值，一半不值”。

值的地方：流式传输确实解决了实际问题，性能提升不是虚的，尤其是数据密集型的场景。跨语言调用也省了不少同步文档和接口约定的精力。

不值的地方：工具链复杂度高了不少，调试成本上升，团队学习曲线明显陡峭。如果只是简单 CRUD 场景，REST 加上 OpenAPI 其实足够用。

技术选型从来不是单纯的技术问题。团队背景、现有基础设施、维护成本，这些都要考虑。如果你的服务已经稳定运行，没有明确的痛点，贸然迁移 gRPC 的ROI可能不高。但如果你正好在搭建新的微服务体系，而且对流式传输、多语言调用、性能有明确需求，gRPC 值得一试。

最后说一句：协议、工具、框架都在变，但解决实际问题的初心不变。别为了"技术正确"而折腾，为了解决问题才折腾。

可用性说明：本文发布于 2021 年 6 月，距今已超过五年。文中涉及的软件版本、接口、下载地址、命令参数和操作界面可能已经发生变化，部分方案在当前环境下可能失效。请结合官方最新文档核对后再操作，生产环境使用前务必先行验证。

版权声明: 本文首发于指尖魔法屋-gRPC折腾手记（https://blog.thinkmoon.cn/post/125-grpc-protocol-practice-guide/）转载或引用必须申明原指尖魔法屋来源及源地址！

配置中心折腾手记

考虑到我们的技术栈是 Spring Cloud，团队对阿里的技术比较熟悉，部署简单也是一个重要因素，最后选了 …

📁 技术实践

📅 2021-05-03

把Docker Swarm换到Kubernetes时踩过的坑

Swarm 跑起来没太大问题，但很快就暴露了一些问题：

📁 技术实践

📅 2022-03-27

微服务通信：同步不够用了之后

最近在项目里又碰了一次微服务通信，有几处判断值得留下来。

📁 技术实践

📅 2022-04-15

gRPC折腾手记

为什么从 REST 换到 gRPC

gRPC 协议其实没那么复杂

实际踩过的几个坑

HTTP/2 依赖不是虚的

消息大小限制吃瘪

超时控制没对齐

序列化兼容性差点翻车

错误处理怎么做才对

监控和调试怎么办

这趟折腾到底值不值

评论

相关阅读