搜索

写给go开发者的gRPC教程-拦截器

Go语言精选

共 11745字，需浏览 24分钟

 · 2023-04-18

本篇为【写给go开发者的gRPC教程】系列第三篇

第一篇：protobuf基础

第二篇：通信模式

第三篇：拦截器

gRPC的拦截器和其他框架的拦截器（也称middleware）作用是一样的。利用拦截器我们可以在不侵入业务逻辑的前提下修改或者记录服务端或客户端的请求与响应，利用拦截器我们可以实现诸如日志记录、权限认证、限流等诸多功能

上一篇提到gRPC的通信模式分为unary和streaming几种模式，拦截器也分为两种：unary interceptors和streaming interceptors ，两种拦截器可以分别应用在服务端和客户端，所以gRPC总共为我们提供了四种拦截器。它们已经被定义成了go中的接口，我们创建的拦截器只要实现这些接口即可

008f98413e43219dbdb5a79ea9c784af.webpgRPC四种拦截器一览

服务端拦截器

服务端的拦截器从请求开始按顺序执行拦截器，在执行完对应RPC的逻辑之后，再按反向的顺序执行拦截器中对响应的处理逻辑

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

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客户端拦截器

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

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 (via AuthFunc) piece of auth middleware

Logging

  • grpc_ctxtags - a library that adds a Tag 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 around grpclog.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 交流学习。


浏览 84
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
了解加密货币到加密货币的互换
1、什么是加密货币互换？加密货币到加密货币的互换是指以现行市场汇率将一种加密货币直接兑换为另一种加密货币。与需要法定货币存款和较长流程的传统交易所不同，加密货币到加密货币的互换可以无缝地促进交换。掉期在提高加密货币的流动性和效率方面发挥着重要作用。该功能使用户能够将他们的加密货币与钱包中的其他代币进
区块链头条
0
李彦宏：开源大模型不如闭源，后者会持续领先；周鸿祎：“开源不如闭源” 的言论是胡说八道
架构师大咖 架构师大咖，打造有价值的架构师交流平台。分享架构师干货、教程、课程、资讯。架构师大咖，每日推送。 公众号该公众号已被封禁0、李彦宏：开源大模型不如闭源，后者会持续领先当今
源码共读
0
【第129期】程序员的新宠：三款终端工具，让你告别Xshell！
概述 WindTerm：跨平台的SSH利器 首先介绍的是WindTerm，这是一款使用C语言开发的跨平台SSH客户端。它不仅完全免费，而且没有商业使用的限制。WindTerm支持SSH v2、Telnet、Raw Tcp等协议，而且性能出色，甚至超过了FinalShell和Electerm。功能
前端微服务
0
字节员工：35岁以后被裁员的，后来都走了哪条路？现在2-2，要不要利用最后一年拼命上个岸。
架构师大咖 架构师大咖，打造有价值的架构师交流平台。分享架构师干货、教程、课程、资讯。架构师大咖，每日推送。 公众号该公众号已被封禁在当今竞争激烈的职场环境中，年龄并不总是一个决定性
源码共读
0
上班的时候，有一群摸鱼搭子非常重要...
上班的时候，有一群摸鱼搭子非常重要！一到上班时间，他们就从四面八方涌进群里冒泡...从八卦聊到股市、从职场聊到乌X兰局势，偶尔还会复读、相亲、battle...然后，下午6点钟准时消失不见...所以你要不要加入我们一起摸鱼？我们有北京、上海、深圳、广州、杭州、武汉、成都、南京等8个城市的摸鱼群，还有
产品经理日记
0
周四002 瑞超：同样落寞的境遇——北雪平vs埃尔夫斯堡
上赛季最终排名联赛第9的北雪平本赛季伊始表现不佳，4轮战罢他们仅以1胜1平2负的战绩排在倒数第三，这支历史上曾夺得13次联赛冠军、6次杯赛冠军老牌劲旅，正如英格兰赛场上的一众百年俱乐部，在低谷中不断探索着出路。球队主教练安德烈亚斯·阿尔姆曾是AIK索尔纳及赫根队的主教练，他于今年年初刚刚拿起球队教鞭
产品与体验
0
雷军辟谣了！不是高考状元，卡里也没有冰冷的 40 亿
架构师大咖 架构师大咖，打造有价值的架构师交流平台。分享架构师干货、教程、课程、资讯。架构师大咖，每日推送。 公众号该公众号已被封禁最近很火的雷军简历，听说落魄时卡里只有冰冷的 40
源码共读
0
日本影山优佳最新杂志照，展现充满透明感的美丽
今天的图文分享的是影山优佳的杂志写真。元日向坂46的影山优佳，登上了写真杂志《周刊FLASH》5/7和‬5/14合并号的封面。影山优佳是日本艺人、女演员、前偶像。身高155厘米。2001年‬5月‬8日‬出生‬于‬东京都。2023年7月从组合日向坂46毕业，之后作为演员活跃的影山优佳，在《周刊FLAS
python教程
0
盘点一个使用超级鹰识别验证码并自动登录的案例
点击上方“Python共享之家”，进行关注回复“资源”即可获赠Python学习资料今日鸡汤江上几人在，天涯孤棹还。大家好，我是皮皮。一、前言前几天在Python钻石交流群【静惜】问了一个Python实现识别验证码并自动登录的问题，提问截图如下：验证码的截图如下所示：二、实现过程这里大家激烈的探讨，【
IT共享之家
0
朋友，你也不想一个人孤孤单单的上班吧？
上班的时候，有一群摸鱼搭子非常重要！一到上班时间，他们就从四面八方涌进群里冒泡...从八卦聊到股市、从职场聊到乌X兰局势，偶尔还会复读、相亲、battle...然后，下午6点钟准时消失不见...所以你要不要加入我们一起摸鱼？我们有北京、上海、深圳、广州、杭州、武汉、成都、南京等8个城市的摸鱼群，还有
产品经理日记
0
点赞
评论
收藏
分享

手机扫一扫分享

举报