写给go开发者的gRPC教程-拦截器
共 11745字,需浏览 24分钟
·
2023-04-18 12:02
本篇为【写给go开发者的gRPC教程】系列第三篇
第一篇:protobuf基础
第二篇:通信模式
第三篇:拦截器
gRPC
的拦截器和其他框架的拦截器(也称middleware)作用是一样的。利用拦截器我们可以在不侵入业务逻辑的前提下修改或者记录服务端或客户端的请求与响应,利用拦截器我们可以实现诸如日志记录、权限认证、限流等诸多功能
上一篇提到gRPC
的通信模式分为unary
和streaming
几种模式,拦截器也分为两种:unary interceptors
和streaming interceptors
,两种拦截器可以分别应用在服务端和客户端,所以gRPC总共为我们提供了四种拦截器。它们已经被定义成了go中的接口,我们创建的拦截器只要实现这些接口即可
![008f98413e43219dbdb5a79ea9c784af.webp](https://filescdn.proginn.com/7123207af8ceca3dd55b3a1bb4bb74ea/008f98413e43219dbdb5a79ea9c784af.webp)
服务端拦截器
服务端的拦截器从请求开始按顺序执行拦截器,在执行完对应RPC的逻辑之后,再按反向的顺序执行拦截器中对响应的处理逻辑
![81327c107ba453066b49a2591d2ff50f.webp](https://filescdn.proginn.com/01ac82ed717ca8b108e756a0bb6e3ad9/81327c107ba453066b49a2591d2ff50f.webp)
unary interceptors
对于unary
服务的拦截器只需实现UnaryServerInterceptor
接口即可
func(ctx context.Context, req interface{},
info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err error)
-
ctx context.Context
:单个请求的上下文 -
req interface{}
:RPC服务的请求结构体 -
info *UnaryServerInfo
:RPC的服务信息 -
handler UnaryHandler
:它包装了服务实现,通过调用它我们可以完成RPC并获取到响应
参数看不懂没关系,我们来看一个例子
示例
// 实现 unary interceptors
func orderUnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
// Pre-processing logic
s := time.Now()
// Invoking the handler to complete the normal execution of a unary RPC.
m, err := handler(ctx, req)
// Post processing logic
log.Printf("Method: %s, req: %s, resp: %s, latency: %s\n",
info.FullMethod, req, m, time.Now().Sub(s))
return m, err
}
func main() {
s := grpc.NewServer(
// 使用 unary interceptors
grpc.UnaryInterceptor(orderUnaryServerInterceptor),
)
pb.RegisterOrderManagementServer(s, &OrderManagementImpl{})
// ...
}
完整代码参考: https://github.com/liangwt/grpc-example/tree /main/06-interceptors/server。下同
假设我们的客户端请求了GetOrder
,根据示例再重新看下拦截器接口的每一个参数
🌲 req interface{}
RPC服务的请求结构体,对于GetOrder
来说就是orderId *wrapperspb.StringValue
🌲 info *UnaryServerInfo
包含两个字段:
FullMethod
是请求的method名字(例如/ecommerce.OrderManagement/getOrder
);
Server
就是服务实现(就是示例RegisterOrderManagementServer
中的&OrderManagementImpl{}
)
🌲 handler
包装了服务实现
所以在调用它之前我们可以进行改写req
或ctx
、记录逻辑开始时间等操作
调用完handler
即完成了RPC并获取到响应,我们不仅可以记录响应还可以改写响应
总结
这张图大致展示了UnaryServerInterceptor
接口的每个参数的含义
![f5e8ffde93f07f589c549aa223a4a52d.webp](https://filescdn.proginn.com/f77c711d386ccf52eb9c72507e0059d9/f5e8ffde93f07f589c549aa223a4a52d.webp)
streaming interceptors
对于stream
服务的拦截器只要实现StreamServerInterceptor
接口即可。它适用于我们上一篇介绍的
- 服务器端流式 RPC
- 客户端流式 RPC
- 双向流式 RPC
func(srv interface{}, ss ServerStream,
info *StreamServerInfo, handler StreamHandler) error
-
srv interface{}
:服务实现 -
ss ServerStream
:服务端视角的流。怎么理解呢?无论是哪一种流式RPC对于服务端来说发送(SendMsg
)就代表着响应数据,接收(RecvMsg
)就代表着请求数据,不同的流式RPC的区别就在于是多次发送数据(服务器端流式 RPC)还是多次接收数据(客户端流式 RPC)或者两者均有(双向流式 RPC)。因此仅使用这一个抽象就代表了所有的流式RPC场景 -
info *StreamServerInfo
:RPC的服务信息 -
handler StreamHandler
:它包装了服务实现,通过调用它我们可以完成RPC
示例
我们来看一个例子
func orderStreamServerInterceptor(srv interface{},
ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
// Pre-processing logic
s := time.Now()
// Invoking the StreamHandler to complete the execution of RPC invocation
err := handler(srv, ss)
// Post processing logic
log.Printf("Method: %s, latency: %s\n", info.FullMethod, time.Now().Sub(s))
return err
}
func main() {
s := grpc.NewServer(
grpc.StreamInterceptor(orderStreamServerInterceptor),
)
pb.RegisterOrderManagementServer(s, &OrderManagementImpl{})
//...
}
根据示例再重新看下拦截器接口的参数
🌲 srv interface{}
服务实现(就是示例RegisterOrderManagementServer
中的&OrderManagementImpl{}
)
🌲 ss grpc.ServerStream
服务端发送和接收数据的接口,注意它是一个接口
🌲 info *grpc.StreamServerInfo
包含三个字段:
FullMethod
是请求的method名字(例如/ecommerce.OrderManagement/updateOrders
);
IsClientStream
是否是客户端流
IsServerStream
是否是服务端流
🌲 handler
包装了服务实现
所以在调用它之前我们可以进行改写数据流、记录逻辑开始时间等操作
调用完handler
即完成了RPC,因为是流式调用所以不会返回响应数据,只有error
流式拦截器既没有请求字段,handler
也不会返回响应,该如何记录、修改请求响应呢?
如果想劫持流数据,答案就在ss ServerStream
。再重复一遍它的含义:服务端视角的流,它是一个接口。无论是哪一种流式RPC对于服务端来说发送(SendMsg
)就代表着响应数据,接收(RecvMsg
)就代表着请求数据,不同的流式RPC的区别就在于是多次发送数据(服务器端流式 RPC)还是多次接收数据(客户端流式 RPC)或者两者均有(双向流式 RPC)。因此可以对ss
进行包装,只要传入handler
的类型实现ServerStream
即可
// SendMsg method call.
type wrappedStream struct {
Recv []interface{}
Send []interface{}
grpc.ServerStream
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
err := w.ServerStream.RecvMsg(m)
w.Recv = append(w.Recv, m)
return err
}
func (w *wrappedStream) SendMsg(m interface{}) error {
err := w.ServerStream.SendMsg(m)
w.Send = append(w.Send, m)
return err
}
func newWrappedStream(s grpc.ServerStream) *wrappedStream {
return &wrappedStream{
make([]interface{}, 0),
make([]interface{}, 0),
s,
}
}
func orderStreamServerInterceptor(srv interface{},
ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
// Pre-processing logic
s := time.Now()
// Invoking the StreamHandler to complete the execution of RPC invocation
nss := newWrappedStream(ss)
err := handler(srv, nss)
// Post processing logic
log.Printf("Method: %s, req: %+v, resp: %+v, latency: %s\n",
info.FullMethod, nss.Recv, nss.Send, time.Now().Sub(s))
return err
}
客户端拦截器
客户端拦截器和服务端拦截器类似,从请求开始按顺序执行拦截器,在获取到服务端响应之后,再按反向的顺序执行拦截器中对响应的处理逻辑
![fb0d524f0f8f40e8f86f74fd9817d721.webp](https://filescdn.proginn.com/3af2815f9f1b1b554ab656b00311da5b/fb0d524f0f8f40e8f86f74fd9817d721.webp)
unary interceptors
client端要实现UnaryClientInterceptor
接口实现的接口如下
func(ctx context.Context, method string, req, reply interface{},
cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error
你可以在调用远程函数前拦截RPC,通过获取RPC相关信息,如参数,上下文,函数名,请求等,你甚至可以修改原始的远程调用
-
ctx context.Context
:单个请求的上下文 -
method string
:请求的method名字(例如/ecommerce.OrderManagement/getOrder
) -
req, reply interface{}
:请求和响应数据 -
cc *ClientConn
:客户端与服务端的链接 -
invoker UnaryInvoker
:通过调用它我们可以完成RPC并获取到响应 -
opts ...CallOption
:RPC调用的所有配置项,包含设置到conn上的,也包含配置在每一个调用上的
示例
func orderUnaryClientInterceptor(ctx context.Context, method string, req, reply interface{},
cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
// Pre-processor phase
s := time.Now()
// Invoking the remote method
err := invoker(ctx, method, req, reply, cc, opts...)
// Post-processor phase
log.Printf("method: %s, req: %s, resp: %s, latency: %s\n",
method, req, reply, time.Now().Sub(s))
return err
}
func main() {
conn, err := grpc.Dial("127.0.0.1:8009",
grpc.WithInsecure(),
grpc.WithUnaryInterceptor(orderUnaryClientInterceptor),
)
if err != nil {
panic(err)
}
c := pb.NewOrderManagementClient(conn)
// ...
}
根据示例再重新看下拦截器接口的参数
🌲 cc *grpc.ClientConn
客户端与服务端的链接
这里的cc
就是示例代码中c := pb.NewOrderManagementClient(conn)
的conn
🌲 invoker grpc.UnaryInvoker
包装了服务实现
调用完invoker
即完成了RPC,所以我们可以改写req
或者在获取到reply
之后修改响应
![5e07fd954553f42ed4fa5dbc6d82cd82.webp](https://filescdn.proginn.com/d4af91220cbe51342d7a2e30ecd4dffd/5e07fd954553f42ed4fa5dbc6d82cd82.webp)
streaming interceptors
要实现的接口StreamClientInterceptor
func(ctx context.Context, desc *StreamDesc, cc *ClientConn,
method string, streamer Streamer, opts ...CallOption) (ClientStream, error)
和serve端类似的参数类似,重点关注下面几个参数
-
cs ClientStream
:客户端视角的流。类比服务端的ss ServerStream
,无论是哪一种流式RPC对于客户端来说发送(SendMsg
)就代表着请求数据,接收(RecvMsg
)就代表着响应数据(正好和服务端是反过来的) -
streamer Streamer
:完成RPC请求的调用
示例
这里不再赘述,可以参考服务端拦截器
func orderStreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc,
cc *grpc.ClientConn, method string, streamer grpc.Streamer,
opts ...grpc.CallOption) (grpc.ClientStream, error) {
// Pre-processing logic
s := time.Now()
cs, err := streamer(ctx, desc, cc, method, opts...)
// Post processing logic
log.Printf("method: %s, latency: %s\n", method, time.Now().Sub(s))
return cs, err
}
func main() {
conn, err := grpc.Dial("127.0.0.1:8009",
grpc.WithInsecure(),
grpc.WithStreamInterceptor(orderStreamClientInterceptor),
)
if err != nil {
panic(err)
}
c := pb.NewOrderManagementClient(conn)
// ...
}
如何记录或者修改流拦截器的请求响应数据?
和服务端stream interceptor
同样的道理,通过包装ClientStream
即可做到
// SendMsg method call.
type wrappedStream struct {
method string
grpc.ClientStream
}
func (w *wrappedStream) RecvMsg(m interface{}) error {
err := w.ClientStream.RecvMsg(m)
log.Printf("method: %s, res: %s\n", w.method, m)
return err
}
func (w *wrappedStream) SendMsg(m interface{}) error {
err := w.ClientStream.SendMsg(m)
log.Printf("method: %s, req: %s\n", w.method, m)
return err
}
func newWrappedStream(method string, s grpc.ClientStream) *wrappedStream {
return &wrappedStream{
method,
s,
}
}
func orderStreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc,
cc *grpc.ClientConn, method string, streamer grpc.Streamer,
opts ...grpc.CallOption) (grpc.ClientStream, error) {
// Pre-processing logic
s := time.Now()
cs, err := streamer(ctx, desc, cc, method, opts...)
// Post processing logic
log.Printf("method: %s, latency: %s\n", method, time.Now().Sub(s))
return newWrappedStream(method, cs), err
}
拦截器链
服务器只能配置一个 unary interceptor
和 stream interceptor
,否则会报错,客户端也是,虽然不会报错,但是只有最后一个才起作用。
// 服务端拦截器
s := grpc.NewServer(
grpc.UnaryInterceptor(orderUnaryServerInterceptor),
grpc.StreamInterceptor(orderStreamServerInterceptor),
)
// 客户端拦截器
conn, err := grpc.Dial("127.0.0.1:8009",
grpc.WithInsecure(),
grpc.WithUnaryInterceptor(orderUnaryClientInterceptor),
grpc.WithStreamInterceptor(orderStreamClientInterceptor),
)
如果你想配置多个,可以使用拦截器链或者自己实现一个。
// 服务端拦截器
s := grpc.NewServer(
grpc.ChainUnaryInterceptor(
orderUnaryServerInterceptor1,
orderUnaryServerInterceptor2,
),
grpc.ChainStreamInterceptor(
orderServerStreamInterceptor1,
orderServerStreamInterceptor2,
),
)
// 客户端拦截器
conn, err := grpc.Dial("127.0.0.1:8009",
grpc.WithInsecure(),
grpc.WithChainUnaryInterceptor(
orderUnaryClientInterceptor1,
orderUnaryClientInterceptor2,
),
grpc.WithChainStreamInterceptor(
orderStreamClientInterceptor1,
orderStreamClientInterceptor2,
),
)
生态
除了可以自己实现拦截器外,gRPC生态也提供了一系列的开源的拦截器可供使用,覆盖权限、日志、监控等诸多方面
https://github.com/grpc-ecosystem/go-grpc-middleware
Auth
-
grpc_auth
- a customizable (viaAuthFunc
) piece of auth middleware
Logging
-
grpc_ctxtags
- a library that adds aTag
map to context, with data populated from request body -
grpc_zap
- integration of zap logging library into gRPC handlers. -
grpc_logrus
- integration of logrus logging library into gRPC handlers. -
grpc_kit
- integration of go-kit/log logging library into gRPC handlers. -
grpc_grpc_logsettable
- a wrapper aroundgrpclog.LoggerV2
that allows to replace loggers in runtime (thread-safe).
Monitoring
-
grpc_prometheus
⚡ - Prometheus client-side and server-side monitoring middleware -
otgrpc
⚡ - OpenTracing client-side and server-side interceptors -
grpc_opentracing
- OpenTracing client-side and server-side interceptors with support for streaming and handler-returned tags
Client
-
grpc_retry
- a generic gRPC response code retry mechanism, client-side middleware
Server
-
grpc_validator
- codegen inbound message validation from.proto
options -
grpc_recovery
- turn panics into gRPC errors -
ratelimit
- grpc rate limiting by your own limiter
示例代码
https://github.com/liangwt/grpc-example/tree/main/06-interceptors
推荐阅读
我为大家整理了一份 从入门到进阶的Go学习资料礼包 ,包含学习建议:入门看什么,进阶看什么。 关注公众号 「polarisxu」,回复 ebook 获取;还可以回复「进群」,和数万 Gopher 交流学习。