gRPC(3):拦截器

在 gRPC 调用过程中,我们可以拦截 RPC 的执行,在 RPC 服务执行前或执行后运行一些自定义逻辑,这在某些场景下很有用,例如身份验证、日志等,我们可以在 RPC 服务执行前检查调用方的身份信息,若未通过验证,则拒绝执行,也可以在执行前后记录下详细的请求响应信息到日志。这种拦截机制与 Gin 中的中间件技术类似,在 gRPC 中被称为 拦截器,它是 gRPC 核心扩展机制之一

拦截器不止可以作用在服务端上,客户端同样可以拦截,在请求发出之前和收到响应之后执行一些自定义逻辑,根据拦截的 RPC 类型,可分为 一元拦截器流拦截器

服务端拦截器

在 gRPC 服务端,可以插入一个或多个拦截器,收到的请求按注册顺序通过各个拦截器,返回响应时则倒序通过:

gRPC(3):拦截器

一元拦截器

通过以下步骤实现一元拦截器:

  • 定义一元拦截器方法:
// 函数名无特殊要求,参数需一致
// 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服务的正常执行
        m, err := handler(ctx, req)
        // 后置处理逻辑
        log.Printf("[unary interceptor resonse] %s", m)
        // 返回响应
        return m, err
}
  • 注册定义的一元拦截器
func main() {
    ...
	// 创建gRPC服务器实例的时候注册拦截器
    // NewServer 可传入多个拦截器
	s := grpc.NewServer(grpc.UnaryInterceptor(orderUnaryServerInterceptor))
    ...
}

流拦截器

流拦截器包括前置处理阶段和流操作阶段,前置处理阶段可以在流 RPC 进入具体服务实现之前进行拦截,而在流操作阶段,可以对流中的每一条消息进行拦截,通过以下步骤实现流拦截器:

  • 自定义一个嵌入grpc.ServerStream的包装器
type wrappedStream struct {
	grpc.ServerStream
}
  • 实现包装器的 RecvMsg 和 SendMsg 方法
// 自定义RecvMsg和SendMsg方法实现对每一个流消息的拦截
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)
}
  • 实现流拦截器
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 调用,客户端同样可以拦截发出去的 RPC 请求以及收到的响应,同样可以实现一元拦截器以及流拦截器:

gRPC(3):拦截器

一元拦截器

和服务端一元拦截器一样的方法,只是方法参数略微有所差别,此外在建立连接的时候注册拦截器,同样可以注册多个拦截器:

// method请求方法字符串,req包含请求的所有信息参数等,reply在实际RPC调用后存储响应信息,通过invoker实际调用
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调用
	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请求发出之前拦截
	log.Printf("[client interceptor send] %s", method)
    // 发出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))
    ...
}

gRPC(3):拦截器

上一篇:程序调试常见问题汇总


下一篇:问题定位小工具