Go语言gRPC微服务实战：Protobuf定义与拦截器中间件设计

小编 • • 后端开发

gRPC是Google基于HTTP/2和Protocol Buffers开发的高性能RPC框架，在微服务架构中被广泛用于内部服务间通信。相比REST+JSON，gRPC使用二进制Protobuf编码，传输体积减少约30%-50%，序列化速度快5-10倍。本文以Go语言实现一套完整的gRPC微服务，涵盖Protobuf定义、服务实现、拦截器中间件、流式通信和健康检查。

Protobuf接口定义与代码生成

Protobuf是gRPC的接口定义语言（IDL），同时作为序列化格式。定义一个用户管理服务的proto文件：

syntax = "proto3";
package user.v1;
option go_package = "github.com/example/proto/user/v1;userv1";

message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  string role = 4;
  int64 created_at = 5;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
  string password = 3;
  string role = 4;
}

message CreateUserResponse {
  User user = 1;
}

message GetUserRequest {
  int64 id = 1;
}

message GetUserResponse {
  User user = 1;
}

message ListUsersRequest {
  int32 page = 1;
  int32 page_size = 2;
  string role = 3;
}

message ListUsersResponse {
  repeated User users = 1;
  int32 total = 2;
}

message StreamUsersRequest {
  string role = 1;
}

service UserService {
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);
  rpc GetUser(GetUserRequest) returns (GetUserResponse);
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
  rpc StreamUsers(StreamUsersRequest) returns (stream User);
}

proto3语法相比proto2更简洁，字段类型默认零值不再区分是否设置。repeated关键字定义数组，stream关键字定义服务端流式响应。option go_package指定生成的Go包路径。

安装protoc工具链并生成Go代码：

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

protoc --go_out=. --go_opt=paths=source_relative \
    --go-grpc_out=. --go-grpc_opt=paths=source_relative \
    proto/user/v1/user.proto

gRPC服务端实现与错误处理

package service

import (
    "context"
    "errors"
    "time"
    userv1 "github.com/example/proto/user/v1"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
)

type UserServiceServer struct {
    userv1.UnimplementedUserServiceServer
    db UserRepo
}

func NewUserServiceServer(db UserRepo) *UserServiceServer {
    return &UserServiceServer{db: db}
}

func (s *UserServiceServer) CreateUser(ctx context.Context, req *userv1.CreateUserRequest) (*userv1.CreateUserResponse, error) {
    if req.GetName() == "" || req.GetEmail() == "" {
        return nil, status.Error(codes.InvalidArgument, "name and email are required")
    }
    
    user, err := s.db.Create(ctx, &User{
        Name:     req.GetName(),
        Email:    req.GetEmail(),
        Role:     req.GetRole(),
        Password: hashPassword(req.GetPassword()),
    })
    if err != nil {
        if errors.Is(err, ErrDuplicateEmail) {
            return nil, status.Error(codes.AlreadyExists, "email already exists")
        }
        return nil, status.Errorf(codes.Internal, "failed to create user: %v", err)
    }
    
    return &userv1.CreateUserResponse{
        User: &userv1.User{
            Id:        user.ID,
            Name:      user.Name,
            Email:     user.Email,
            Role:      user.Role,
            CreatedAt: user.CreatedAt.Unix(),
        },
    }, nil
}

func (s *UserServiceServer) GetUser(ctx context.Context, req *userv1.GetUserRequest) (*userv1.GetUserResponse, error) {
    user, err := s.db.FindByID(ctx, req.GetId())
    if err != nil {
        if errors.Is(err, ErrNotFound) {
            return nil, status.Errorf(codes.NotFound, "user %d not found", req.GetId())
        }
        return nil, status.Error(codes.Internal, err.Error())
    }
    
    return &userv1.GetUserResponse{User: toProtoUser(user)}, nil
}

// 服务端流式RPC：批量推送用户数据
func (s *UserServiceServer) StreamUsers(req *userv1.StreamUsersRequest, stream userv1.UserService_StreamUsersServer) error {
    users, err := s.db.FindByRole(context.Background(), req.GetRole())
    if err != nil {
        return status.Error(codes.Internal, err.Error())
    }
    
    for _, user := range users {
        select {
        case <-stream.Context().Done():
            return stream.Context().Err()
        default:
            if err := stream.Send(toProtoUser(user)); err != nil {
                return err
            }
            time.Sleep(100 * time.Millisecond)
        }
    }
    return nil
}

gRPC错误处理使用status包，通过codes定义标准错误码（InvalidArgument、NotFound、Internal等），客户端通过status.FromError提取错误信息。stream.Send在循环中发送数据，stream.Context().Done()检测客户端是否断开连接。

拦截器中间件：日志、认证与链路追踪

gRPC拦截器类似Web框架的中间件，分为Unary拦截器（一元RPC）和Stream拦截器（流式RPC）。以下实现一组通用拦截器：

package interceptor

import (
    "context"
    "log/slog"
    "time"
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
)

// 一元拦截器：请求日志
func UnaryLoggingInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
        start := time.Now()
        resp, err = handler(ctx, req)
        duration := time.Since(start)
        st, _ := status.FromError(err)
        
        logger.Info("gRPC request",
            "method", info.FullMethod,
            "code", st.Code().String(),
            "duration_ms", duration.Milliseconds(),
        )
        return resp, err
    }
}

// 一元拦截器：JWT认证
func UnaryAuthInterceptor(publicKey []byte) grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        if isPublicMethod(info.FullMethod) {
            return handler(ctx, req)
        }
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, status.Error(codes.Unauthenticated, "metadata is missing")
        }
        tokens := md.Get("authorization")
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "token is missing")
        }
        claims, err := validateJWT(tokens[0], publicKey)
        if err != nil {
            return nil, status.Error(codes.Unauthenticated, "invalid token")
        }
        ctx = context.WithValue(ctx, userClaimsKey{}, claims)
        return handler(ctx, req)
    }
}

// 流式拦截器：日志
func StreamLoggingInterceptor(logger *slog.Logger) grpc.StreamServerInterceptor {
    return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
        start := time.Now()
        err := handler(srv, ss)
        duration := time.Since(start)
        st, _ := status.FromError(err)
        logger.Info("gRPC stream",
            "method", info.FullMethod,
            "code", st.Code().String(),
            "duration_ms", duration.Milliseconds(),
        )
        return err
    }
}

func isPublicMethod(method string) bool {
    return method == "/grpc.health.v1.Health/Check" || 
           method == "/grpc.reflection.v1.ServerReflectionInfo"
}

拦截器链按注册顺序执行。auth拦截器在logging之后，确保认证失败也能记录日志。metadata.FromIncomingContext从gRPC请求头提取键值对，客户端通过metadata.AppendToOutgoingContext注入认证信息。

服务端启动与健康检查

package main

import (
    "log/slog"
    "net"
    "google.golang.org/grpc"
    "google.golang.org/grpc/health"
    healthpb "google.golang.org/grpc/health/grpc_health_v1"
    "google.golang.org/grpc/reflection"
    userv1 "github.com/example/proto/user/v1"
)

func main() {
    logger := slog.Default()
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        logger.Error("failed to listen", "error", err)
        return
    }
    
    grpcServer := grpc.NewServer(
        grpc.ChainUnaryInterceptor(
            interceptor.UnaryLoggingInterceptor(logger),
            interceptor.UnaryAuthInterceptor(pubKey),
        ),
        grpc.ChainStreamInterceptor(
            interceptor.StreamLoggingInterceptor(logger),
        ),
    )
    
    userSvc := service.NewUserServiceServer(db)
    userv1.RegisterUserServiceServer(grpcServer, userSvc)
    
    healthSvc := health.NewServer()
    healthSvc.SetServingStatus("user.v1.UserService", healthpb.HealthCheckResponse_SERVING)
    healthpb.RegisterHealthServer(grpcServer, healthSvc)
    
    reflection.Register(grpcServer)
    
    logger.Info("gRPC server starting", "addr", lis.Addr().String())
    if err := grpcServer.Serve(lis); err != nil {
        logger.Error("failed to serve", "error", err)
    }
}

health服务遵循gRPC健康检查协议，Kubernetes中可配置grpc-health-probe作为liveness探针。reflection服务允许grpcurl等工具在不预编译proto的情况下直接调用，仅限开发环境开启。

客户端连接池与负载均衡

package client

import (
    "context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/metadata"
    userv1 "github.com/example/proto/user/v1"
)

type UserClient struct {
    conn   *grpc.ClientConn
    client userv1.UserServiceClient
}

func NewUserClient(addr string) (*UserClient, error) {
    conn, err := grpc.Dial(addr,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithDefaultServiceConfig(`{"loadBalancingConfig":[{"round_robin":{}}]}`),
    )
    if err != nil {
        return nil, err
    }
    return &UserClient{
        conn:   conn,
        client: userv1.NewUserServiceClient(conn),
    }, nil
}

func (c *UserClient) CreateUser(ctx context.Context, token, name, email string) (*userv1.User, error) {
    ctx = metadata.AppendToOutgoingContext(ctx, "authorization", token)
    resp, err := c.client.CreateUser(ctx, &userv1.CreateUserRequest{
        Name:  name,
        Email: email,
        Role:  "editor",
    })
    if err != nil {
        return nil, fmt.Errorf("create user failed: %w", err)
    }
    return resp.GetUser(), nil
}

grpc.Dial创建到服务端的连接，连接可复用且自动重连。WithDefaultServiceConfig配置round_robin负载均衡策略，配合DNS解析多个后端地址实现客户端侧负载均衡。生产环境应使用credentials.NewTLS配置传输加密。

服务治理方面，gRPC内置重试机制通过service config配置，可按方法名设置不同的重试策略。消息中间件（如Kafka）处理gRPC不适合的异步场景，gRPC保持同步通信的定位不变。业务中台建设中gRPC适合内部高性能调用，对外API网关层将gRPC转换为REST供前端使用。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/go-yu-yan-grpc-wei-fu-wu-shi-zhan-protobuf-ding-yi-yu-lan/

赞 (0)
小编小编
0
上一篇 10小时前
下一篇 10小时前

相关推荐