在 gRPC 調(diào)用過程中，我們可以攔截 RPC 的執(zhí)行，在 RPC 服務執(zhí)行前或執(zhí)行后運行一些自定義邏輯，這在某些場景下很有用，例如身份驗證、日志等，我們可以在 RPC 服務執(zhí)行前檢查調(diào)用方的身份信息，若未通過驗證，則拒絕執(zhí)行，也可以在執(zhí)行前后記錄下詳細的請求響應信息到日志。這種攔截機制與 Gin 中的中間件技術類似，在 gRPC 中被稱為 攔截器，它是 gRPC 核心擴展機制之一

攔截器不止可以作用在服務端上，客戶端同樣可以攔截，在請求發(fā)出之前和收到響應之后執(zhí)行一些自定義邏輯，根據(jù)攔截的 RPC 類型，可分為 一元攔截器 和 流攔截器

服務端攔截器

在 gRPC 服務端，可以插入一個或多個攔截器，收到的請求按注冊順序通過各個攔截器，返回響應時則倒序通過：

未命名文件

一元攔截器

通過以下步驟實現(xiàn)一元攔截器：

• 定義一元攔截器方法：

// 函數(shù)名無特殊要求，參數(shù)需一致
// req包含請求的所有信息，info包含一元RPC服務的所有信息
func orderUnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler) (interface{}, error) {
        // 前置處理邏輯
        log.Printf("[unary interceptor request] %s", info.FullMethod)
        // 完成RPC服務的正常執(zhí)行
        m, err := handler(ctx, req)
        // 后置處理邏輯
        log.Printf("[unary interceptor resonse] %s", m)
        // 返回響應
        return m, err
}

• 注冊定義的一元攔截器

func main() {
    ...
    // 創(chuàng)建gRPC服務器實例的時候注冊攔截器
    // NewServer 可傳入多個攔截器
    s := grpc.NewServer(grpc.UnaryInterceptor(orderUnaryServerInterceptor))
    ...
}

流攔截器

流攔截器包括前置處理階段和流操作階段，前置處理階段可以在流 RPC 進入具體服務實現(xiàn)之前進行攔截，而在流操作階段，可以對流中的每一條消息進行攔截，通過以下步驟實現(xiàn)流攔截器：

• 自定義一個嵌入grpc.ServerStream的包裝器

type wrappedStream struct {
    grpc.ServerStream
}

• 實現(xiàn)包裝器的 RecvMsg 和 SendMsg 方法

// 自定義RecvMsg和SendMsg方法實現(xiàn)對每一個流消息的攔截
func (w *wrappedStream) RecvMsg(m interface{}) error {
    log.Printf("[stream interceptor recv] type: %T", m)
    return w.ServerStream.RecvMsg(m)
}
func (w *wrappedStream) SendMsg(m interface{}) error {
    log.Printf("[stream interceptor send] %s", m)
    return w.ServerStream.SendMsg(m)
}

• 實現(xiàn)流攔截器

func orderServerStreamInterceptor(srv interface{}, ss grpc.ServerStream,
    info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    // 前置處理階段
    log.Printf("[stream interceptor request] %s", info.FullMethod)
    // 使用自定義包裝器處理流
    err := handler(srv, &wrappedStream{ss})
    if err != nil {
        log.Printf("[stream Intercept error] %v", err)
    }
    return err
}

• 注冊流攔截器

func main() {
    ...
    s := grpc.NewServer(grpc.StreamInterceptor(orderServerStreamInterceptor))
    ...
}

客戶端攔截器

在服務端可以攔截收到的 RPC 調(diào)用，客戶端同樣可以攔截發(fā)出去的 RPC 請求以及收到的響應，同樣可以實現(xiàn)一元攔截器以及流攔截器：

未命名文件 (1)

一元攔截器

和服務端一元攔截器一樣的方法，只是方法參數(shù)略微有所差別，此外在建立連接的時候注冊攔截器，同樣可以注冊多個攔截器：

// method請求方法字符串，req包含請求的所有信息參數(shù)等，reply在實際RPC調(diào)用后存儲響應信息，通過invoker實際調(diào)用
func orderUnaryClientInterceptor(ctx context.Context, method string, req, reply interface{},
    cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
    // 前置處理階段
    log.Println("method: " + method)
    // 實際的RPC調(diào)用
    err := invoker(ctx, method, req, reply, cc, opts...)
    // 后置處理
    log.Println(reply)
    return err
}

func main() {
    ...
    conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithUnaryInterceptor(orderUnaryClientInterceptor))
    ...
}

流攔截器

流攔截器也是和服務端一樣的步驟：

type wrappedStream struct {
    grpc.ClientStream
}

func (w *wrappedStream) SendMsg(m interface{}) error {
    log.Printf("[stream interceptor send] %s", m)
    return w.ClientStream.SendMsg(m)
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
    log.Printf("[stream interceptor recv] type: %T", m)
    return w.ClientStream.RecvMsg(m)
}

func orderClientStreamInterceptor(ctx context.Context, desc *grpc.StreamDesc,
    cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
    // 前置處理階段，RPC請求發(fā)出之前攔截
    log.Printf("[client interceptor send] %s", method)
    // 發(fā)出RPC請求
    s, err := streamer(ctx, desc, cc, method, opts...)
    if err != nil {
        return nil, err
    }
    return &wrappedStream{s}, nil
}

func main() {
    ...
    conn, err := grpc.Dial(address, grpc.WithInsecure(),
        grpc.WithStreamInterceptor(orderClientStreamInterceptor))
    ...
}