乐趣区

关于grpc:grpc中的拦截器

0.1、索引

https://waterflow.link/articles/1665853719750

当咱们编写 HTTP 应用程序时,您能够应用 HTTP 中间件包装特定于路由的应用程序处理程序,能够在执行应用程序处理程序之前和之后执行一些常见的逻辑。咱们通常应用中间件来编写跨畛域组件,例如受权、日志记录、缓存等。在 gRPC 中能够应用称为拦截器的概念来实现雷同的性能。

通过应用拦截器,咱们能够在客户端和服务器上拦挡 RPC 办法的执行。在客户端和服务器上,都有两种类型的拦截器:

  • UnaryInterceptor(一元拦截器)
  • StreamInterceptor(流式拦截器)

UnaryInterceptor 拦挡一元 RPC,而 StreamInterceptor 拦挡流式 RPC。

在一元 RPC 中,客户端向服务器发送单个申请并返回单个响应。在流式 RPC 中,客户端或服务器,或单方(双向流式传输),获取一个流读取一系列音讯返回,而后客户端或服务器从返回的流中读音讯,直到没有更多音讯为止。

1、在 gRPC 客户端中编写拦截器

咱们能够在 gRPC 客户端应用程序中编写两种类型的拦截器:

  • UnaryClientInterceptor:UnaryClientInterceptor 拦挡客户端上一元 RPC 的执行。
  • StreamClientInterceptor:StreamClientInterceptor 拦挡 ClientStream 的创立。它可能会返回一个自定义的 ClientStream 来拦挡所有 I/O 操作。

1、UnaryClientInterceptor

为了创立 UnaryClientInterceptor,能够通过提供 UnaryClientInterceptor 函数值调用 WithUnaryInterceptor 函数,该函数返回一个 grpc.DialOption 指定一元 RPC 的拦截器:

func WithUnaryInterceptor(f UnaryClientInterceptor) DialOption

而后将返回的 grpc.DialOption 值用作调用 grpc.Dial 函数以将拦截器利用于一元 RPC 的参数。

UnaryClientInterceptor func 类型的定义如下:

type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts …CallOption) error

参数调用者是实现 RPC 的处理程序,调用它是拦截器的责任。UnaryClientInterceptor 函数值提供拦截器逻辑。这是一个实现 UnaryClientInterceptor 的示例拦截器:

clientInterceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {start := time.Now().Unix()
        err := invoker(ctx, method, req, reply, cc, opts...)
        end := time.Now().Unix()
    // 获取调用 grpc 的申请时长
        fmt.Println("request time duration:", end - start)
        return err
    }

上面的函数返回一个 grpc.DialOption 值,它通过提供 UnaryClientInterceptor 函数值来调用 WithUnaryInterceptor 函数:

func WithUnaryInterceptorCustom() grpc.DialOption {clientInterceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {start := time.Now().Unix()
        err := invoker(ctx, method, req, reply, cc, opts...)
        end := time.Now().Unix()
    // 获取调用 grpc 的申请时长
        fmt.Println("request time duration:", end - start)
        return err
    }
    return grpc.WithUnaryInterceptor(clientInterceptor)
}

返回的 grpc.DialOption 值用作调用 grpc.Dial 函数以利用拦截器的参数:

conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure(), WithUnaryInterceptorCustom())

搭建简略 grpc 服务能够参考这篇文章:https://waterflow.link/articl…

2、StreamClientInterceptor

为了创立 StreamClientInterceptor,能够通过提供 StreamClientInterceptor 函数值调用 WithStreamInterceptor 函数,该函数返回一个 grpc.DialOption 指定流 RPC 的拦截器:

func WithStreamInterceptor(f StreamClientInterceptor) DialOption {return newFuncDialOption(func(o *dialOptions) {o.streamInt = f})
}

而后将返回的 grpc.DialOption 值用作调用 grpc.Dial 函数的参数,以将拦截器利用于流式 RPC。

上面是 StreamClientInterceptor func 类型的定义:

type StreamClientInterceptor func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error)

上面是 StreamClientInterceptor 的具体实现:

// 包装下 stream
// 构造体内嵌接口,初始化的时候须要赋值对象实现了该接口的所有办法
type wrappedStream struct {grpc.ClientStream}

// 实现接管音讯办法,并自定义打印
func (w *wrappedStream) RecvMsg(m interface{}) error  {log.Printf("====== [Client Stream Interceptor]" +
        "Receive a message (Type: %T) at %v",
        m, time.Now().Format(time.RFC3339))
    return w.ClientStream.RecvMsg(m)
}

// 实现发送音讯办法,并自定义打印
func (w *wrappedStream) SendMsg(m interface{}) error {log.Printf("====== [Client Stream Interceptor]" +
        "Send a message (Type: %T) at %v",
        m, time.Now().Format(time.RFC3339))
    return w.ClientStream.SendMsg(m)
}

// 初始化包装 stream
func newWrappedStream(s grpc.ClientStream) grpc.ClientStream {return &wrappedStream{s}
}

func WithStreamInterceptorCustom() grpc.DialOption {clientInterceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {clientStream, err := streamer(ctx, desc, cc, method, opts...)
        if err != nil {return nil, err}
    // 返回包装后的 stream
    // 这里 clientStream 实现了 grpc.ClientStream 接口
        return newWrappedStream(clientStream), err
    }
    return grpc.WithStreamInterceptor(clientInterceptor)
}

这里须要留神:

  1. 定义一个包装 stream 构造体 wrappedStream,这里用到了构造体内嵌接口的形式,间接实现了接口的所有办法,具体能够看正文
  2. 重写 RecvMsg 和 SendMsg 办法
  3. WithStreamInterceptorCustom 拦截器中染回包装后的 clientStream

为了将 StreamClientInterceptor 利用于流式 RPC,只需将 WithStreamInterceptor 函数返回的 grpc.DialOption 值作为调用 grpc.Dial 函数的参数传递。您能够将 UnaryClientInterceptor 和 StreamClientInterceptor 值传递给 grpc.Dial 函数。

conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure(), WithUnaryInterceptorCustom(), WithStreamInterceptorCustom())

残缺的客户端代码像,上面这样:

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "grpcdemo/helloservice"
    "io"
    "log"
    "time"
)

func main() {
    // 连贯 grpc 服务端,退出拦截器
    conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure(), WithUnaryInterceptorCustom(), WithStreamInterceptorCustom())
    if err != nil {log.Fatal(err)
    }
    defer conn.Close()

  // 一元 rpc
    unaryRpc(conn)
  // 流式 rpc
    streamRpc(conn)

}

// 一元拦截器
func WithUnaryInterceptorCustom() grpc.DialOption {clientInterceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {start := time.Now().Unix()
        err := invoker(ctx, method, req, reply, cc, opts...)
        end := time.Now().Unix()
        fmt.Println("invoker request time duration:", end - start)
        return err
    }
    return grpc.WithUnaryInterceptor(clientInterceptor)
}

type wrappedStream struct {grpc.ClientStream}

func (w *wrappedStream) RecvMsg(m interface{}) error  {log.Printf("====== [Client Stream Interceptor]" +
        "Receive a message (Type: %T) at %v",
        m, time.Now().Format(time.RFC3339))
    return w.ClientStream.RecvMsg(m)
}

func (w *wrappedStream) SendMsg(m interface{}) error {log.Printf("====== [Client Stream Interceptor]" +
        "Send a message (Type: %T) at %v",
        m, time.Now().Format(time.RFC3339))
    return w.ClientStream.SendMsg(m)
}

func newWrappedStream(s grpc.ClientStream) grpc.ClientStream {return &wrappedStream{s}
}

// 流式拦截器
func WithStreamInterceptorCustom() grpc.DialOption {clientInterceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {clientStream, err := streamer(ctx, desc, cc, method, opts...)
        if err != nil {return nil, err}
        return newWrappedStream(clientStream), err
    }
    return grpc.WithStreamInterceptor(clientInterceptor)
}

func unaryRpc(conn *grpc.ClientConn) {client := helloservice.NewHelloServiceClient(conn)
    reply, err := client.Hello(context.Background(), &helloservice.String{Value: "hello"})
    if err != nil {log.Fatal(err)
    }
    log.Println("unaryRpc recv:", reply.Value)
}

func streamRpc(conn *grpc.ClientConn) {client := helloservice.NewHelloServiceClient(conn)
    stream, err := client.Channel(context.Background())
    if err != nil {log.Fatal(err)
    }

    go func() {
        for {if err := stream.Send(&helloservice.String{Value: "hi"}); err != nil {log.Fatal(err)
            }
            time.Sleep(time.Second)
        }
    }()

    for {recv, err := stream.Recv()
        if err != nil {
            if err == io.EOF {break}
            log.Fatal(err)
        }

        fmt.Println("streamRpc recv:", recv.Value)

    }
}

能够联合我之前的文章,把本期的代码加进去运行调试

(搭建简略 grpc 服务能够参考这篇文章:https://waterflow.link/articl…)

运行成果如下:

go run helloclient/main.go
invoker request time duration:  1
2022/10/14 23:17:35 unaryRpc recv:  hello:hello
2022/10/14 23:17:35 ====== [Client Stream Interceptor] Receive a message (Type: *helloservice.String) at 2022-10-14T23:17:35+08:00
2022/10/14 23:17:35 ====== [Client Stream Interceptor] Send a message (Type: *helloservice.String) at 2022-10-14T23:17:35+08:00
2022/10/14 23:17:36 ====== [Client Stream Interceptor] Send a message (Type: *helloservice.String) at 2022-10-14T23:17:36+08:00
streamRpc recv:  hello:hi
2022/10/14 23:17:36 ====== [Client Stream Interceptor] Receive a message (Type: *helloservice.String) at 2022-10-14T23:17:36+08:00
2022/10/14 23:17:37 ====== [Client Stream Interceptor] Send a message (Type: *helloservice.String) at 2022-10-14T23:17:37+08:00
streamRpc recv:  hello:hi
2022/10/14 23:17:37 ====== [Client Stream Interceptor] Receive a message (Type: *helloservice.String) at 2022-10-14T23:17:37+08:00
2022/10/14 23:17:38 ====== [Client Stream Interceptor] Send a message (Type: *helloservice.String) at 2022-10-14T23:17:38+08:00
streamRpc recv:  hello:hi
2022/10/14 23:17:38 ====== [Client Stream Interceptor] Receive a message (Type: *helloservice.String) at 2022-10-14T23:17:38+08:00

2、在 gRPC 客户端中编写拦截器

和 gRPC 客户端应用程序一样,gRPC 服务器应用程序提供两种类型的拦截器:

  • UnaryServerInterceptor:提供了一个钩子来拦挡服务器上一元 RPC 的执行。
  • StreamServerInterceptor:提供了一个钩子来拦挡服务器上流式 RPC 的执行。

1、UnaryServerInterceptor

为了创立 UnaryServerInterceptor,能够通过提供 UnaryServerInterceptor 函数值作为参数调用 UnaryInterceptor 函数,该参数返回为服务器设置 UnaryServerInterceptor 的 grpc.ServerOption 值。

func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {return newFuncServerOption(func(o *serverOptions) {
        if o.unaryInt != nil {panic("The unary server interceptor was already set and may not be reset.")
        }
        o.unaryInt = i
    })
}

而后应用返回的 grpc.ServerOption 值作为参数提供给 grpc.NewServer 函数以注册为 UnaryServerInterceptor。

UnaryServerInterceptor 函数的定义如下:

func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error)

参数 info 蕴含了这个 RPC 的所有信息,拦截器能够对其进行操作。而 handler 是服务办法实现的包装器。拦截器负责调用处理程序来实现 RPC。

1、定义一个服务端的鉴权一元拦截器
func ServerUnaryInterceptorCustom() grpc.ServerOption {serverInterceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {start := time.Now()

    // 如果是非登录申请,须要验证 token
        if info.FullMethod != "/helloservice.HelloService/Login" {if err := authorize(ctx); err != nil {return nil, err}
        }

        h, err := handler(ctx, req)

        log.Printf("Request - Method:%s\tDuration:%s\tError:%v\n",
            info.FullMethod,
            time.Since(start),
            err)
        return h, err
    }
    return grpc.UnaryInterceptor(serverInterceptor)
}

// authorize 从 Metadata 中获取 token 并校验是否非法
func authorize(ctx context.Context) error {
  // 从 context 中获取 metadata
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {return status.Errorf(codes.InvalidArgument, "Retrieving metadata is failed")
    }

    authHeader, ok := md["authorization"]
    if !ok {return status.Errorf(codes.Unauthenticated, "Authorization token is not supplied")
    }

    token := authHeader[0]
    // validateToken function validates the token
    err := validateToken(token)

    if err != nil {return status.Errorf(codes.Unauthenticated, err.Error())
    }
    return nil
}

func validateToken(token string) error {
    // 校验 token
    return nil
}

咱们能够看下咱们定义的一元拦截器的执行流程:

  1. 首先进来之后咱们判断如果是登录申请,间接转发申请,并打印日志
  2. 如果是非登录申请,须要验证 token,调用 authorize 办法
  3. 在 authorize 办法中,会从 context 中获取 metadata 元数据,而后解析获取 token 并验证

请留神,后面代码块中的拦截器逻辑应用包 google.golang.org/grpc/codes 和 google.golang.org/grpc/status。

2、grpc 客户端传入 token

gRPC 反对在客户端和服务器之间应用 Context 值发送元数据。google.golang.org/grpc/metadata 包提供了元数据的性能。

其中 MD 类型是一个 k - v 的 map,想上面这样

type MD map[string][]string

上面咱们在客户端编写向服务端发送 token 的代码,咱们批改下客户端的 unaryRpc,结构蕴含 authorization 的 metadata:

func unaryRpc(conn *grpc.ClientConn) {client := helloservice.NewHelloServiceClient(conn)
    ctx := context.Background()
  // 结构元数据,并返回 MD 类型的构造
    md := metadata.Pairs("authorization", "mytoken")
  // 元数据塞入 context 并返回新的 context
    ctx = metadata.NewOutgoingContext(ctx, md)
    reply, err := client.Hello(ctx, &helloservice.String{Value: "hello"})
    if err != nil {log.Fatal(err)
    }
    log.Println("unaryRpc recv:", reply.Value)
}

这样元数据的信息就会跟着 context 发送到 grpc 服务端。

接着咱们在服务端 grpc 中批改如下代码,退出一行日志:

func validateToken(token string) error {log.Printf("get the token: %s \n", token)
    // 校验 token
    return nil
}
3、运行服务

咱们从新执行下 grpc 服务端程序,而后运行下客户端代码,能够看到 token 传过来了:

go run helloservice/main/main.go
2022/10/15 20:36:04 server started...
2022/10/15 20:36:08 get the token: mytoken 
2022/10/15 20:36:09 Request - Method:/helloservice.HelloService/Hello   Duration:1.001216763s   Error:<nil>

2、StreamServerInterceptor

为了创立 StreamServerInterceptor,通过提供 StreamServerInterceptor func 值作为参数调用 StreamInterceptor 函数,该参数返回为服务器设置 StreamServerInterceptor 的 grpc.ServerOption 值。

func StreamInterceptor(i StreamServerInterceptor) ServerOption {return newFuncServerOption(func(o *serverOptions) {
        if o.streamInt != nil {panic("The stream server interceptor was already set and may not be reset.")
        }
        o.streamInt = i
    })
}

而后应用返回的 grpc.ServerOption 值作为参数提供给 grpc.NewServer 函数以注册为 UnaryServerInterceptor。

上面是 StreamServerInterceptor func 类型的定义:

type StreamServerInterceptor func(srv interface{}, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error

咱们看下服务端流式拦截器的具体例子:

type wrappedStream struct {grpc.ServerStream}

func (w *wrappedStream) RecvMsg(m interface{}) error  {log.Printf("====== [Server Stream Interceptor]" +
        "Receive a message (Type: %T) at %v",
        m, time.Now().Format(time.RFC3339))
    return w.ServerStream.RecvMsg(m)
}

func (w *wrappedStream) SendMsg(m interface{}) error {log.Printf("====== [Server Stream Interceptor]" +
        "Send a message (Type: %T) at %v",
        m, time.Now().Format(time.RFC3339))
    return w.ServerStream.SendMsg(m)
}

func newWrappedStream(s grpc.ServerStream) grpc.ServerStream {return &wrappedStream{s}
}

func ServerStreamInterceptorCustom() grpc.ServerOption {serverInterceptor := func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {return handler(srv, newWrappedStream(ss))
    }
    return grpc.StreamInterceptor(serverInterceptor)
}

下面服务端流式拦截器代码可参考客户端流式拦截器的代码,根本差不多。

3、多拦截器

go-grpc 在 v1.28.0 之前是不反对多个拦截器。然而能够应用一些第三方的包,拦截器链接允利用多个拦截器。

v1.28.0 之后曾经能够反对多个拦截器,咱们批改下服务端代码如下:

...

unaryInterceptors := []grpc.ServerOption {ServerUnaryInterceptorCustom(),
        ServerStreamInterceptorCustom(),}
grpcServer := grpc.NewServer(unaryInterceptors...)

...
退出移动版