gRPC微服务实战：从协议设计到生产踩坑-深圳市維司達科技有限公司

为什么选gRPC

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

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

后来换成gRPC：

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

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

基础概念

gRPC有四种通信模式：

Unary：一问一答，最常用
Server Streaming：客户端一个请求，服务端返回流
Client Streaming：客户端发送流，服务端一个响应
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; }

几个关键点

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

用buf管理Proto

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

# buf.yamlversion:v1breaking:use:-FILElint:use:-DEFAULTexcept:-PACKAGE_VERSION_SUFFIXdeps:-buf.build/googleapis/googleapis

# buf.gen.yamlversion:v1managed:enabled:truego_package_prefix:default:github.com/example/gen/goplugins:-plugin:buf.build/protocolbuffers/goout:gen/goopt:paths=source_relative-plugin:buf.build/grpc/goout:gen/goopt:paths=source_relative

生成代码：

buf generate

Go服务端实现

packagemainimport("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")typeuserServerstruct{userv1.UnimplementedUserServiceServer repo UserRepository}func(s*userServer)GetUser(ctx context.Context,req*userv1.GetUserRequest)(*userv1.GetUserResponse,error){ifreq.UserId<=0{returnnil,status.Error(codes.InvalidArgument,"user_id is required")}user,err:=s.repo.GetByID(ctx,req.UserId)iferr!=nil{iferrors.Is(err,ErrNotFound){returnnil,status.Error(codes.NotFound,"user not found")}returnnil,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){iflen(req.UserIds)==0{returnnil,status.Error(codes.InvalidArgument,"user_ids is required")}iflen(req.UserIds)>100{returnnil,status.Error(codes.InvalidArgument,"too many user_ids, max 100")}users,err:=s.repo.BatchGetByIDs(ctx,req.UserIds)iferr!=nil{returnnil,status.Error(codes.Internal,"internal error")}protoUsers:=make([]*userv1.User,0,len(users))for_,u:=rangeusers{protoUsers=append(protoUsers,toProtoUser(u))}return&userv1.BatchGetUsersResponse{Users:protoUsers,},nil}funcmain(){lis,err:=net.Listen("tcp",":50051")iferr!=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),// 4MBgrpc.MaxSendMsgSize(4*1024*1024),}server:=grpc.NewServer(opts...)userv1.RegisterUserServiceServer(server,&userServer{})log.Println("gRPC server listening on :50051")iferr:=server.Serve(lis);err!=nil{log.Fatalf("failed to serve: %v",err)}}

Go客户端实现

packagemainimport("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")funcmain(){// 客户端配置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...)iferr!=nil{log.Fatalf("failed to dial: %v",err)}deferconn.Close()client:=userv1.NewUserServiceClient(conn)// 调用示例ctx,cancel:=context.WithTimeout(context.Background(),5*time.Second)defercancel()resp,err:=client.GetUser(ctx,&userv1.GetUserRequest{UserId:12345,})iferr!=nil{log.Fatalf("GetUser failed: %v",err)}log.Printf("User: %v",resp.User)}

拦截器实现

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

服务端拦截器

// 日志拦截器funcLoggingInterceptor(ctx context.Context,reqinterface{},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)returnresp,err}// 恢复拦截器（防止panic导致服务挂掉）funcRecoveryInterceptor(ctx context.Context,reqinterface{},info*grpc.UnaryServerInfo,handler grpc.UnaryHandler)(respinterface{},errerror){deferfunc(){ifr:=recover();r!=nil{log.Printf("panic recovered: %v\n%s",r,debug.Stack())err=status.Error(codes.Internal,"internal error")}}()returnhandler(ctx,req)}// 认证拦截器funcAuthInterceptor(ctx context.Context,reqinterface{},info*grpc.UnaryServerInfo,handler grpc.UnaryHandler)(interface{},error){md,ok:=metadata.FromIncomingContext(ctx)if!ok{returnnil,status.Error(codes.Unauthenticated,"missing metadata")}tokens:=md.Get("authorization")iflen(tokens)==0{returnnil,status.Error(codes.Unauthenticated,"missing token")}userID,err:=validateToken(tokens[0])iferr!=nil{returnnil,status.Error(codes.Unauthenticated,"invalid token")}// 把用户ID放到context里ctx=context.WithValue(ctx,"user_id",userID)returnhandler(ctx,req)}// 使用server:=grpc.NewServer(grpc.ChainUnaryInterceptor(RecoveryInterceptor,LoggingInterceptor,AuthInterceptor,),)

客户端拦截器

// 超时拦截器funcTimeoutInterceptor(timeout time.Duration)grpc.UnaryClientInterceptor{returnfunc(ctx context.Context,methodstring,req,replyinterface{},cc*grpc.ClientConn,invoker grpc.UnaryInvoker,opts...grpc.CallOption)error{ctx,cancel:=context.WithTimeout(ctx,timeout)defercancel()returninvoker(ctx,method,req,reply,cc,opts...)}}// 重试拦截器funcRetryInterceptor(maxRetriesint)grpc.UnaryClientInterceptor{returnfunc(ctx context.Context,methodstring,req,replyinterface{},cc*grpc.ClientConn,invoker grpc.UnaryInvoker,opts...grpc.CallOption)error{varlastErrerrorfori:=0;i<maxRetries;i++{err:=invoker(ctx,method,req,reply,cc,opts...)iferr==nil{returnnil}// 只重试可重试的错误if!isRetryable(err){returnerr}lastErr=err time.Sleep(time.Duration(i+1)*100*time.Millisecond)}returnlastErr}}funcisRetryable(errerror)bool{s,ok:=status.FromError(err)if!ok{returnfalse}switchs.Code(){casecodes.Unavailable,codes.ResourceExhausted,codes.Aborted:returntruedefault:returnfalse}}

流式通信

实时数据推送场景用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)defers.subscribers.Delete(userID)for{select{case<-stream.Context().Done():returnnilcasenotification:=<-ch:iferr:=stream.Send(notification);err!=nil{returnerr}}}}

客户端：

stream,err:=client.Subscribe(ctx,&notificationv1.SubscribeRequest{UserId:12345,})iferr!=nil{log.Fatal(err)}for{notification,err:=stream.Recv()iferr==io.EOF{break}iferr!=nil{log.Fatal(err)}log.Printf("Received: %v",notification)}

生产环境配置

健康检查

import"google.golang.org/grpc/health"importhealthpb"google.golang.org/grpc/health/grpc_health_v1"healthServer:=health.NewServer()healthpb.RegisterHealthServer(server,healthServer)// 设置服务健康状态healthServer.SetServingStatus("user.v1.UserService",healthpb.HealthCheckResponse_SERVING)

优雅关闭

funcmain(){server:=grpc.NewServer()// 注册服务...// 优雅关闭quit:=make(chanos.Signal,1)signal.Notify(quit,syscall.SIGINT,syscall.SIGTERM)gofunc(){<-quit log.Println("shutting down server...")// 停止接收新连接，等待现有请求完成server.GracefulStop()}()iferr:=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"// 注册自定义resolverresolver.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（杀鸡焉用牛刀）

实践建议：

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

gRPC微服务实战：从协议设计到生产踩坑