gRPC 实战：内部通信的协议之争

// proto/order/v1/order.proto
syntax = "proto3";

package order.v1;

option go_package = "github.com/example/gen/go/order/v1;orderv1";

// 订单状态枚举
enum OrderStatus {
    ORDER_STATUS_UNSPECIFIED = 0;
    ORDER_STATUS_CREATED = 1;
    ORDER_STATUS_PAID = 2;
    ORDER_STATUS_SHIPPED = 3;
    ORDER_STATUS_COMPLETED = 4;
}

// 订单消息
message Order {
    string id = 1;
    string customer_id = 2;
    repeated OrderItem items = 3;
    OrderStatus status = 4;
    int64 created_at = 5; // Unix timestamp
}

message OrderItem {
    string product_id = 1;
    int32 quantity = 2;
    int64 unit_price_cents = 3; // 用分避免浮点误差
}

// 创建订单请求
message CreateOrderRequest {
    string customer_id = 1;
    repeated OrderItem items = 2;
}

message CreateOrderResponse {
    Order order = 1;
}

// 订单服务定义
service OrderService {
    // 一元调用：创建订单
    rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);

    // 服务端流式：订阅订单状态变更
    rpc WatchOrder(WatchOrderRequest) returns (stream OrderStatusEvent);
}

// 订阅请求
message WatchOrderRequest {
    string order_id = 1;
}

// 状态变更事件
message OrderStatusEvent {
    string order_id = 1;
    OrderStatus status = 2;
    int64 occurred_at = 3;
}

// cmd/order-server/main.go
package main

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

    "google.golang.org/grpc"
    "google.golang.org/grpc/reflection"

    orderv1 "github.com/example/gen/go/order/v1"
)

type orderServer struct {
    orderv1.UnimplementedOrderServiceServer
    // 注入仓储、消息总线等依赖
}

// CreateOrder 处理下单请求
func (s *orderServer) CreateOrder(ctx context.Context, req *orderv1.CreateOrderRequest) (*orderv1.CreateOrderResponse, error) {
    if len(req.Items) == 0 {
        return nil, status.Error(codes.InvalidArgument, "订单必须包含至少一个商品")
    }

    // 调用领域服务创建订单（省略业务逻辑）
    order := &orderv1.Order{
        Id:         generateID(),
        CustomerId: req.CustomerId,
        Items:      req.Items,
        Status:     orderv1.OrderStatus_ORDER_STATUS_CREATED,
        CreatedAt:  time.Now().Unix(),
    }

    return &orderv1.CreateOrderResponse{Order: order}, nil
}

// WatchOrder 服务端流式：推送订单状态变更
func (s *orderServer) WatchOrder(req *orderv1.WatchOrderRequest, stream orderv1.OrderService_WatchOrderServer) error {
    // 模拟状态变更事件流
    ticker := time.NewTicker(2 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-stream.Context().Done():
            return stream.Context().Err()
        case t := <-ticker.C:
            event := &orderv1.OrderStatusEvent{
                OrderId:    req.OrderId,
                Status:     orderv1.OrderStatus_ORDER_STATUS_PAID,
                OccurredAt: t.Unix(),
            }
            if err := stream.Send(event); err != nil {
                return err
            }
        }
    }
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("监听失败: %v", err)
    }

    grpcServer := grpc.NewServer(
        grpc.UnaryInterceptor(orderLoggingInterceptor),
    )
    orderv1.RegisterOrderServiceServer(grpcServer, &orderServer{})

    // 开启反射服务，方便 grpcurl 调试
    reflection.Register(grpcServer)

    log.Printf("gRPC server listening on :50051")
    if err := grpcServer.Serve(lis); err != nil {
        log.Fatalf("服务异常退出: %v", err)
    }
}

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

# 调用方法
grpcurl -plaintext -d '{"customer_id":"u-001","items":[{"product_id":"p-1","quantity":1,"unit_price_cents":1000}]}' \
    localhost:50051 order.v1.OrderService/CreateOrder

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

// 注册健康检查服务
healthServer := health.NewServer()
healthpb.RegisterHealthServer(grpcServer, healthServer)

// 服务注册到 gRPC Server 后，立即标记为 SERVING
healthServer.SetServingStatus("order.v1.OrderService", healthpb.HealthCheckResponse_SERVING)

// 在应用优雅退出前，标记为 NOT_SERVING（让 LB 不再发新请求）
healthServer.SetServingStatus("order.v1.OrderService", healthpb.HealthCheckResponse_NOT_SERVING)

// 监听系统信号，触发退出流程
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
<-sigCh
log.Println("收到终止信号，停止接收新请求...")
grpcServer.GracefulStop()

readinessProbe:
  grpc:
    port: 50051
  initialDelaySeconds: 10
  periodSeconds: 5

go func() {
    <-sigCh
    log.Println("开始优雅停止，等待 30 秒...")
    done := make(chan struct{})
    go func() {
        grpcServer.GracefulStop()
        close(done)
    }()
    select {
    case <-done:
        log.Println("优雅退出完成")
    case <-time.After(30 * time.Second):
        log.Println("优雅退出超时，强制停止")
        grpcServer.Stop()
    }
}()

// internal/order-client/client.go
package orderclient

import (
    "context"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"

    orderv1 "github.com/example/gen/go/order/v1"
)

type Client struct {
    conn   *grpc.ClientConn
    client orderv1.OrderServiceClient
}

func New(addr string) (*Client, error) {
    conn, err := grpc.Dial(addr,
        grpc.WithInsecure(),
        grpc.WithBlock(),
        grpc.WithTimeout(5*time.Second),
    )
    if err != nil {
        return nil, err
    }
    return &Client{
        conn:   conn,
        client: orderv1.NewOrderServiceClient(conn),
    }, nil
}

// CreateOrderWithTraceID 携带追踪上下文的调用
func (c *Client) CreateOrderWithTraceID(ctx context.Context, req *orderv1.CreateOrderRequest, traceID string) (*orderv1.CreateOrderResponse, error) {
    // 把 traceID 注入 gRPC metadata（服务端可从 metadata 取出）
    md := metadata.Pairs("x-trace-id", traceID)
    ctx = metadata.NewOutgoingContext(ctx, md)

    // 单次调用 3 秒超时
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()

    resp, err := c.client.CreateOrder(ctx, req)
    if err != nil {
        st, _ := status.FromError(err)
        return nil, fmt.Errorf("order service error: code=%s msg=%s", st.Code(), st.Message())
    }
    return resp, nil
}

func (c *Client) Close() error {
    return c.conn.Close()
}

// pkg/interceptor/logging.go
package interceptor

import (
    "context"
    "log"
    "time"

    "google.golang.org/grpc"
)

// UnaryLoggingInterceptor 一元调用日志拦截器
func UnaryLoggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    start := time.Now()

    // 从 context 提取 trace 信息（由 tracing 拦截器注入）
    traceID, _ := ctx.Value("traceID").(string)

    resp, err := handler(ctx, req)

    duration := time.Since(start)
    if err != nil {
        log.Printf("[%s] %s FAILED duration=%v err=%v", err, info.FullMethod, duration, err)
    } else {
        log.Printf("[%s] %s OK duration=%v", info.FullMethod, duration, traceID)
    }
    return resp, err
}

// 注册到 gRPC Server
grpcServer := grpc.NewServer(
    grpc.UnaryInterceptor(UnaryLoggingInterceptor),
)

message Order {
    string id = 1;
    string customer_id = 2;
    // 永远不要删字段或改字段编号
    reserved 3; // 标记为废弃字段
    reserved "old_field";
}

// 全局连接池
var (
    orderConn *grpc.ClientConn
    payConn   *grpc.ClientConn
)

func init() {
    orderConn, _ = grpc.Dial("order-service:50051",
        grpc.WithInsecure(),
        grpc.WithBalancerName("round_robin"),  // 客户端负载均衡
    )
    payConn, _ = grpc.Dial("pay-service:50051",
        grpc.WithInsecure(),
    )
}

// 在 handler 中使用
orderClient := orderv1.NewOrderServiceClient(orderConn)
resp, err := orderClient.CreateOrder(ctx, req)

import (
    "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
    "github.com/grpc-ecosystem/go-grpc-middleware"
    "github.com/uber/jaeger-client-go"
)

tracer, closer := jaeger.NewTracer(...) // 省略初始化
defer closer.Close()

grpcServer := grpc.NewServer(
    grpc.UnaryInterceptor(
        grpc_middleware.ChainUnaryServer(
            opentracing.UnaryServerInterceptor(tracer),
            UnaryLoggingInterceptor,
        ),
    ),
)