GRPC (3) : 拦截器
上篇文章记录了 gRPC 四种通信模式及实现,今天学习一个新的知识点——拦截器。通常客户端请求到达服务端的时候不会立即进行业务处理,而是进行一些预处理操作,比如监控数据采集(统计 QPS),链路追踪,身份信息校验,必传参数校验等等。gRPC 为此提供了一个拦截器(Interceptor)功能来实现这一系列的操作。按照通信方式可以分为一元拦截器(Unary Interceptor)和流拦截器(Streaming Interceptor),按照应用角色可以分为客户端拦截器(Client-Side Interceptor)和服务端拦截器(Server-Side Interceptor),具体类型如下
// grpc interceptor.go
grpc.UnaryClientInterceptor
grpc.UnaryServerInterceptor
grpc.StreamClientInterceptor
grpc.StreamServerInterceptor
服务端拦截器(Server-Side Interceptor)
一元拦截器(Unary Interceptor)
type UnaryServerInterceptor func(
ctx context.Context, // 请求上下文,可以做一些超时处理
req interface{}, // gRPC 请求参数
info *UnaryServerInfo, // gRPC 服务接口信息
handler UnaryHandler, // gRPC 实际调用方法
) (resp interface{}, err error)
流拦截器(Streaming Interceptor)
type StreamServerInterceptor func(
srv interface{}, // 请求参数
ss ServerStream, // gRPC 服务端流信息
info *StreamServerInfo, // gRPC 服务接口信息
handler StreamHandler // gRPC 实际调用方法
) error
客户端拦截器(Client-Side Interceptor)
一元拦截器(Unary Interceptor)
type UnaryClientInterceptor func(
ctx context.Context, // 请求上下文,可以做一些超时处理
method string, // 请求方法
req, reply interface{}, // 请求和响应
cc *ClientConn, // 连接信息
invoker UnaryInvoker, // 调用的 gRPC 方法
opts …CallOption // gRPC 调用预处理接口活后处理接口
) error
流拦截器(Streaming Interceptor)
type StreamClientInterceptor func(
ctx context.Context, // 请求上下文
desc *StreamDesc, // 调用 gRPC 方法流信息
cc *ClientConn, // 连接信息
method string, // 调用方法
streamer Streamer, // 流对象,通过 desc 初始化
opts …CallOption // gRPC 调用预处理接口活后处理接口
) (ClientStream, error)
拦截器(interceptor)处理过程
拦截器的处理过程可以分为以下三个阶段:
预处理阶段(pre-processing)
调用 RPC 方法(invoking RPC method)
后处理阶段(post-processing)
客户端一元拦截器(Client-Side Unary Interceptor)
func mySqlUnaryClientInterceptor(
ctx context.Context,
method string,
req, reply interface{},
cc *grpc.ClientConn,
invoker grpc.UnaryInvoker,
opts …grpc.CallOption) error {
// Pre-processing logic
// set requestID
uuID, _ := uuid.NewRandom()
ctx = context.WithValue(ctx, “requestID”, uuID)
start := time.Now()
log.Printf(“client unary interceptor pre-processing: requstID [%s]\n”,
ctx.Value(“requestID”))
// Invoking the remote method
err := invoker(ctx, method, req, reply, cc, opts…)
// Post processing logic
end := time.Now()
log.Printf(“client unary interceptor post processing: time [%s]\n”,
end.Sub(start))
return err
}
服务端一元拦截器(Server-Side Unary Interceptor)
// mySqlUnaryServerInterceptor 服务端一元拦截器
func mySqlUnaryServerInterceptor(
ctx context.Context,
req interface{},
info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler) (interface{}, error) {
// Pre-processing logic
// set requestID
uuID, _ := uuid.NewRandom()
ctx = context.WithValue(ctx, “requestID”, uuID)
start := time.Now()
log.Printf(“server unary interceptor pre-processing: requstID [%s]\n”,
ctx.Value(“requestID”))
// Invoking the handler to complete the normal execution of a unary RPC.
m, err := handler(ctx, req)
// Post processing logic
end := time.Now()
log.Printf(“server unary interceptor post processing: time [%s]\n”,
end.Sub(start))
return m, err
}
流拦截器预处理阶段和与一元拦截器类似,调用 RPC 方法和后处理两个阶段则不相同,stream 都是 streamer (客户端:clientStream, 服务端:serverStream)调用 SendMsg 和 RecvMsg 获取的,streamer 又是调用 RPC 方法获取的,因此在流拦截器中我们可以对 Streamer 进行封装,在封装的方法 SendMsg 实现流的预处理和 RecvMsg 实现流的拦截
客户端流拦截器(Client-Side Stream Interceptor)
// wrappedStream 用于包装 grpc.ClientStream 结构体并拦截其对应的方法。
type wrappedStream struct {
grpc.ClientStream
}
func newWrappedStream(s grpc.ClientStream) grpc.ClientStream {
return &wrappedStream{s}
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
log.Printf(“Client Stream Interceptor Post Processing RecvMsg (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 Post Processing SendMsg (Type: %T) at %v”,
m, time.Now().Format(time.RFC3339))
return w.ClientStream.SendMsg(m)
}
func mysqlClientStreamInterceptor(
ctx context.Context, desc *grpc.StreamDesc,
cc *grpc.ClientConn, method string,
streamer grpc.Streamer, opts …grpc.CallOption) (grpc.ClientStream, error) {
// Pre-processing
log.Println(“Client Stream Interceptor Pre-processing : “, method)
s, err := streamer(ctx, desc, cc, method, opts…)
if err != nil {
return nil, err
}
return newWrappedStream(s), nil
}
服务端流拦截器(Server-Side Stream Interceptor)
type wrappedStream struct {
grpc.ServerStream
}
func newWrappedStream(s grpc.ServerStream) grpc.ServerStream {
return &wrappedStream{s}
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
log.Printf(“Server Stream Interceptor Post Processing RecvMsg (Type: %T) at %s”,
m, time.Now().Format(time.RFC3339))
return w.ServerStream.RecvMsg(m)
}
func (w *wrappedStream) SendMsg(m interface{}) error {
log.Printf(“Server Stream Interceptor Post Processing SendMsg (Type: %T) at %v”,
m, time.Now().Format(time.RFC3339))
return w.ServerStream.SendMsg(m)
}
func mysqlServerStreamInterceptor(
srv interface{}, ss grpc.ServerStream,
info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
// Pre-processing
log.Println(“Server Stream Interceptor Pre-processing : “, info.FullMethod)
// Invoking the StreamHandler to complete the execution of RPC invocation
err := handler(srv, newWrappedStream(ss))
if err != nil {
log.Printf(“RPC failed with error %v”, err)
}
return err
}
使用方法
我们可以在调用 RPC 方法之前和之后添加自己的逻辑,比如链路追踪设置,请求耗时数据采集等等。客户端和服务端调用方法如下所示:
// 客户端添加一元拦截器
conn, err := grpc.Dial(address, grpc.WithInsecure(),
grpc.WithUnaryInterceptor(mySqlUnaryClientInterceptor),
grpc.WithStreamInterceptor(clientStreamInterceptor))
// 服务端添加一元拦截器
s := grpc.NewServer(
grpc.UnaryInterceptor(mySqlUnaryServerInterceptor),
grpc.StreamInterceptor(orderServerStreamInterceptor))
示例代码执行结果
客户端一元拦截器(Client-Side Unary Interceptor)
2022/07/03 23:09:01 client unary interceptor pre-processing: requstID [d01970fb-96a7-463f-ad68-d9ac597e8e34]
2022/07/03 23:09:02 client unary interceptor post processing: time [142.6873ms]
2022/07/03 23:09:02 select record success []
服务端一元拦截器(Server-Side Unary Interceptor)
2022/07/03 23:09:02 server unary interceptor pre-processing: requstID [ea90a302-9fe0-404d-b309-998638df91ca]
2022/07/03 23:09:02 receives a select record request with sql : SELECT * FROM users WHERE id=1;
2022/07/03 23:09:02 server unary interceptor post processing: time [115.369ms]
客户端流拦截器(Client-Side Stream Interceptor)
2022/07/03 23:01:33 Client Stream Interceptor Pre-processing : /_02_grpc_demo.MysqlService/UpdateRecord
2022/07/03 23:01:33 Client Stream Interceptor Post Processing SendMsg (Type: *pb.MysqlReq) at 2022-07-03T23:01:33+08:00
2022/07/03 23:01:33 Client Stream Interceptor Post Processing RecvMsg (Type: *pb.MysqlRes) at 2022-07-03T23:01:33+08:00
2022/07/03 23:01:33 received response: code:200 msg:”update records sql [UPDATE users SET role = \”stu+tech\” WHERE name = \”hamming_test\”] succ”
2022/07/03 23:01:33 Client Stream Interceptor Post Processing RecvMsg (Type: *pb.MysqlRes) at 2022-07-03T23:01:33+08:00
2022/07/03 23:01:33 no more responses
服务端流拦截器(Server-Side Stream Interceptor)
2022/07/03 23:01:33 Server Stream Interceptor Pre-processing : /_02_grpc_demo.MysqlService/UpdateRecord
2022/07/03 23:01:33 Server Stream Interceptor Post Processing RecvMsg (Type: *pb.MysqlReq) at 2022-07-03T23:01:33+08:00
2022/07/03 23:01:33 Server Stream Interceptor Post Processing RecvMsg (Type: *pb.MysqlReq) at 2022-07-03T23:01:33+08:00
2022/07/03 23:01:33 read ends
2022/07/03 23:01:33 receives a update record request with sql : UPDATE users SET role = “stu+tech” WHERE name = “hamming_test”
2022/07/03 23:01:33 Server Stream Interceptor Post Processing SendMsg (Type: *pb.MysqlRes) at