一. 拦截器的基础使用
- 在go-zero 中提供了拦截器功能
- go-zero中拦截器可以在两个角度分类
- 类型角度分为: 一元拦截器, 流式拦截
- 服务角度分为: 服务端拦截器, 客户端拦截器
- 什么是一元拦截器, 流式拦截
- 一元拦截器是指拦截一元RPC调用的拦截器，一元RPC调用是指客户端发送一个请求，服务端返回一个响应的调用。一元拦截器可以在请求和响应之间执行一些逻辑，例如日志、认证、限流等。一元拦截器是一个函数类型，它接收一个上下文、一个方法名、一个请求、一个响应、一个客户端连接、一个调用器和一些调用选项作为参数，返回一个错误作为结果
- 流式拦截器是指拦截流式RPC调用的拦截器，流式RPC调用是指客户端和服务端可以互相发送多个消息的调用。流式拦截器可以在流开始时执行一些逻辑，也可以在每个消息发送或接收时执行一些逻辑，例如日志、认证、限流等。流式拦截器是一个函数类型，它接收一个上下文、一个流描述、一个客户端连接、一个方法名、一个流创建器和一些调用选项作为参数，返回一个客户端流和一个错误作为结果
1. 服务端拦截器
- 代码示例
- 编写服务端一元拦截器函数, 流式拦截器函数
- 将一元拦截器,流式拦截器注册到服务Server中
import (
"context"
"errors"
"flag"
"fmt"
"github.com/zeromicro/go-zero/core/logx"
"google.golang.org/grpc/metadata"
"log"
"time"
"go_cloud_demo/rpc/internal/config"
"go_cloud_demo/rpc/internal/server"
"go_cloud_demo/rpc/internal/svc"
"go_cloud_demo/rpc/types/user"
"github.com/zeromicro/go-zero/core/conf"
"github.com/zeromicro/go-zero/core/service"
"github.com/zeromicro/go-zero/zrpc"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
)
// 命令行参数读取配置文件所在路径
var configFile = flag.String("f", "rpc/etc/user.yaml", "the config file")
func main() {
flag.Parse()
//1.读取配置文件解析到Config结构体上
var c config.Config
conf.MustLoad(*configFile, &c)
//2.创房服务运行上下文
ctx := svc.NewServiceContext(c)
//3.将服务注册到rpc服务器,并且监听指定端口启动服务
//参数一"c.RpcServerConf":保存了当前rpc服务配置信息
//参数二"func(grpcServer *grpc.Server)"一个函数,当执行该函数时
//会调用通过proto生成的RegisterXXXServer(),将当前rpc服务实现注册到rpc服务器
s := zrpc.MustNewServer(c.RpcServerConf,
func(grpcServer *grpc.Server) {
user.RegisterUserServer(grpcServer, server.NewUserServer(ctx))
if c.Mode == service.DevMode || c.Mode == service.TestMode {
reflection.Register(grpcServer)
}
})
//添加UnaryInterceptor一元拦截器
s.AddUnaryInterceptors(interceptor)
//添加StreamInterceptor流式拦截器
s.AddStreamInterceptors(StreamLoggerInterceptor)
defer s.Stop()
fmt.Printf("Starting rpc server at %s...\n", c.ListenOn)
s.Start()
}
// 自定义一元拦截器函数
func interceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
// 获取metadata 这里的metadata类似http的header,不合法直接return
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, errors.New("获取metadata失败")
}
if values, ok := md["header"]; ok {
// ...
fmt.Printf("接收请求头: %v", values)
}
logx.Info("拦截器前...")
// 记录开始时间和请求的方法
log.Println("start:" + time.Now().Format("2006-01-02 15:04:05") + " " + info.FullMethod)
resp, err := handler(ctx, req)
logx.Info("拦截器后...")
// 正常结束，记录结束时间和方法
log.Println("end:" + time.Now().Format("2006-01-02 15:04:05") + " " + info.FullMethod)
return resp, err
}
// 自定义流式拦截器函数
func StreamLoggerInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
logx.Info("拦截器前...")
log.Println(time.Now().Format("2006-01-02 15:04:05") + " " + info.FullMethod)
err := handler(srv, ss)
logx.Info("拦截器后...")
if err != nil {
log.Println(time.Now().Format("2006-01-02 15:04:05") + " " + err.Error())
return err
}
log.Println(time.Now().Format("2006-01-02 15:04:05") + " " + info.FullMethod)
return nil
}
2. 客户端拦截器
- 编写客户端一元拦截器,流式拦截器函数
import (
"context"
"github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/zrpc"
"go_cloud_demo/rpc/types/user"
"go_cloud_demo/rpc/userclient"
"go_cloud_demo/user/internal/config"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
type ServiceContext struct {
Config config.Config
//用来创建rpc客户端的结构体
RpcUser userclient.User
//访问rpc服务接口返回的数据
UserAuthResp *user.UserAuthResp
}
func NewServiceContext(c config.Config) *ServiceContext {
zrpc.WithUnaryClientInterceptor(interceptor)
return &ServiceContext{
Config: c,
//添加初始化rpc客户端逻辑
//注册客户端拦截器
RpcUser: userclient.NewUser(zrpc.MustNewClient(
c.RpcClientConf,
zrpc.WithUnaryClientInterceptor(interceptor), //添加一元拦截
zrpc.WithStreamClientInterceptor(ClientStreamLoggerInterceptor), //添加流式拦截器
)),
}
}
// 客户端一元拦截器函数
func interceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
md := metadata.New(map[string]string{"name": "lsz"})
ctx = metadata.NewOutgoingContext(ctx, md)
logx.Info("调用rpc服务前")
err := invoker(ctx, method, req, reply, cc)
if err != nil {
return err
}
logx.Info("调用rpc服务后")
return nil
}
// 客户端流式拦截器函数(示例,函数内部为空,不能实际使用)
func ClientStreamLoggerInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
return nil, nil
}
二. 拦截器底层底层执行原理
三. go-zero默认添加的拦截器
客户端
- 在使用 go-zero rpc 客户端时, 调用github.com/zeromicro/go-zero/zrp下的MustNewClient()初始化rpc客户端,示例代码
func NewServiceContext(c config.Config) *ServiceContext {
zrpc.WithUnaryClientInterceptor(interceptor)
return &ServiceContext{
Config: c,
//添加初始化rpc客户端逻辑
RpcUser: userclient.NewUser(zrpc.MustNewClient(
c.RpcClientConf,
zrpc.WithUnaryClientInterceptor(interceptor), //添加自定义一元拦截
zrpc.WithStreamClientInterceptor(ClientStreamLoggerInterceptor), //添加自定义流式拦截器
)),
}
}
- 查看MustNewClient()源码,内部的调用链路上,最终会调用到client结构体上的buildDialOptions()方法
MustNewClient()
--->NewClient()
--->github.com/zeromicro/go-zero/zrpc/internal下的NewClient()
--->client结构体上的dial()
--->client结构体上的buildDialOptions()
//可以参考查看"go 进阶 go-zero相关: 五. 服务发现" 文档
- 查看client结构体上buildDialOptions()源码, 通过该方法默认注册了6个拦截器
func (c *client) buildDialOptions(opts ...ClientOption) []grpc.DialOption {
var cliOpts ClientOptions
for _, opt := range opts {
opt(&cliOpts)
}
var options []grpc.DialOption
if !cliOpts.Secure {
options = append([]grpc.DialOption(nil), grpc.WithTransportCredentials(insecure.NewCredentials()))
}
if !cliOpts.NonBlock {
options = append(options, grpc.WithBlock())
}
options = append(options,
//五个一元拦截器
WithUnaryClientInterceptors(
clientinterceptors.UnaryTracingInterceptor,
clientinterceptors.DurationInterceptor,
clientinterceptors.PrometheusInterceptor,
clientinterceptors.BreakerInterceptor, //熔断器相关拦截器
clientinterceptors.TimeoutInterceptor(cliOpts.Timeout),
),
//一个流式拦截器
WithStreamClientInterceptors(
clientinterceptors.StreamTracingInterceptor,
),
)
return append(options, cliOpts.DialOptions...)
}
// WithStreamClientInterceptors uses given client stream interceptors.
func WithStreamClientInterceptors(interceptors ...grpc.StreamClientInterceptor) grpc.DialOption {
return grpc.WithChainStreamInterceptor(interceptors...)
}
// WithUnaryClientInterceptors uses given client unary interceptors.
func WithUnaryClientInterceptors(interceptors ...grpc.UnaryClientInterceptor) grpc.DialOption {
return grpc.WithChainUnaryInterceptor(interceptors...)
}
- 接下来我们重点看一下熔断器拦截器
1. 熔断器拦截器 BreakerInterceptor
- zRPC中熔断器的实现参考了Google Sre过载保护算法，该算法的原理如下
- 请求数量(requests)：调用方发起请求的数量总和
- 请求接受数量(accepts)：被调用方正常处理的请求数量
- 正常情况下这两个值是相等的,随着被调用方服务出现异常开始拒绝请求,请求接受数量(accepts)的值开始逐渐小于请求数量(requests)，这个时候调用方可以继续发送请求，直到requests = K * accepts，一旦超过这个限制，熔断器就回打开，新的请求会在本地以一定的概率被抛弃直接返回错误,通过修改算法中的K(倍值)，可以调节熔断器的敏感度，当降低该倍值会使自适应熔断算法更敏感，当增加该倍值会使得自适应熔断算法降低敏感度，举例来说，假设将调用方的请求上限从 requests = 2 * acceptst 调整为 requests = 1.1 * accepts 那么就意味着调用方每十个请求之中就有一个请求会触发熔断
- 服务调用方为每一个调用服务(调用路径)维护一个状态机,熔断器的三种状态:
- 关闭(Closed)：该状态下,需要一个计数器来记录调用失败的次数和总的请求次数,如果在某个时间窗口内,失败的失败率达到预设的阈值,则切换到断开状态,此时开启一个超时时间,当到达该时间则切换到半关闭状态,该超时时间是给了系统一次机会来修正导致调用失败的错误,以回到正常的工作状态,在关闭状态下,调用错误是基于时间的,在特定的时间间隔内会重置,这能够防止偶然错误导致熔断器进去断开状态
- 打开(Open)：该状态下,发起请求时会立即返回错误,一般会启动一个超时计时器,当计时器超时后,状态切换到半打开状态,也可以设置一个定时器,定期的探测服务是否恢复
- 半打开(Half-Open)：该状态下,允许应用程序一定数量的请求发往被调用服务,如果这些调用正常,可以认为被调用服务已经恢复正常,熔断器切换到关闭状态,同时重置计数,如果仍有调用失败的情况,则认为被调用方仍然没有恢复,熔断器会切换到关闭状态,然后重置计数器,半打开状态能够有效防止正在恢复中的服务被突然大量请求再次打垮
- 查看熔断拦截器BreakerInterceptor()源码
- 基于请求方法进行熔断,所以该函数中首先会拼接拦截器名target+method
- 然后执行github.com/zeromicro/go-zero/core/breaker下的DoWithAcceptable()函数
//在github.com/zeromicro/go-zero/zrpc/internal/clientinterceptors下
func BreakerInterceptor(ctx context.Context, method string, req, reply interface{},
cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
//1.基于请求方法进行熔断,所以拼接拦截器名target+method
breakerName := path.Join(cc.Target(), method)
return breaker.DoWithAcceptable(
breakerName,
func() error {//发起一次grpc请求函数
return invoker(ctx, method, req, reply, cc, opts...)
},
codes.Acceptable //定义哪些错误码为需要拦截的
)
}
//定义哪些错误码为需要拦截的函数,在github.com/zeromicro/go-zero/zrpc/internal/codes下
//用来判断哪些error会计入失败计数
func Acceptable(err error) bool {
switch status.Code(err) {
case codes.DeadlineExceeded, codes.Internal, codes.Unavailable, codes.DataLoss, codes.Unimplemented:
return false
default:
return true
}
}
- 查看github.com/zeromicro/go-zero/core/breaker下的DoWithAcceptable()函数,该函数执行需要三个参数
- name: 拦截器名称
- req func() error: 发起一次实际grpc请求函数
- acceptable Acceptable: 返回哪些代码需要拦截的函数
func DoWithAcceptable(name string, req func() error, acceptable Acceptable) error {
//执行获取拦截器函数do()
return do(name, func(b Breaker) error {
//最终执行Breaker下的DoWithAcceptable()方法
return b.DoWithAcceptable(req, acceptable)
})
}
//获取拦截器
func do(name string, execute func(b Breaker) error) error {
return execute(GetBreaker(name))
}
// GetBreaker returns the Breaker with the given name.
func GetBreaker(name string) Breaker {
lock.RLock()
b, ok := breakers[name]
lock.RUnlock()
if ok {
return b
}
lock.Lock()
b, ok = breakers[name]
if !ok {
b = NewBreaker(WithName(name))
breakers[name] = b
}
lock.Unlock()
return b
}
- 查看Breaker下的DoWithAcceptable()方法,Breaker是一个接口,默认情况下执行circuitBreaker实现的DoWithAcceptable(),该方法中重点关注调用执行了circuitBreaker下throttle的doReq()方法, 总结就是:
go-zero默认情况下针对这个Breaker接口提供了circuitBreaker实现,执行该结构体上的doReq()方法, 通过执行该方法最终执行到googleBreaker结构体上的doReq(), googleBreaker才是熔断的核心
func (cb *circuitBreaker) DoWithAcceptable(req func() error, acceptable Acceptable) error {
//执行circuitBreaker下throttle的doReq()方法
return cb.throttle.doReq(req, nil, acceptable)
}
- throttle是一个接口内部有allow与doReq两个抽象方法,在上方执行时,默认情况下会执行loggedThrottle这个实现类的
throttle interface {
allow() (Promise, error)
doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error
}
//实现上方throttle 接口
type loggedThrottle struct {
name string
internalThrottle
errWin *errorWindow
}
func newLoggedThrottle(name string, t internalThrottle) loggedThrottle {
return loggedThrottle{
name: name,
internalThrottle: t,
errWin: new(errorWindow),
}
}
func (lt loggedThrottle) allow() (Promise, error) {
promise, err := lt.internalThrottle.allow()
return promiseWithReason{
promise: promise,
errWin: lt.errWin,
}, lt.logError(err)
}
//拦截器执行的方法
func (lt loggedThrottle) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
//重点关注会获取loggedThrottle上的internalThrottle属性,执行它的doReq()方法
return lt.logError(lt.internalThrottle.doReq(req, fallback, func(err error) bool {
accept := acceptable(err)
if !accept && err != nil {
lt.errWin.add(err.Error())
}
return accept
}))
}
- 查看googleBreaker是针对internalThrottle接口的实现, go-zero通过它提供了默认的熔断逻辑
type googleBreaker struct {
k float64 //倍值 默认1.5
stat *collection.RollingWindow //滑动时间窗口，用来对请求失败和成功计数
proba *mathx.Proba //动态概率
}
func newGoogleBreaker() *googleBreaker {
bucketDuration := time.Duration(int64(window) / int64(buckets))
st := collection.NewRollingWindow(buckets, bucketDuration)
return &googleBreaker{
stat: st,
k: k,
proba: mathx.NewProba(),
}
}
//上方会调用该方法
func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
//执行accept(),判断是否触发熔断
if err := b.accept(); err != nil {
if fallback != nil {
return fallback(err)
}
return err
}
defer func() {
if e := recover(); e != nil {
b.markFailure()
panic(e)
}
}()
//执行真正的调用
err := req()
//正常请求计数
if acceptable(err) {
//实际执行：b.stat.Add(1)
//也就是说：内部指标统计成功+1
b.markSuccess()
} else {
//异常请求计数
//原理同上
b.markFailure()
}
return err
}
func (b *googleBreaker) accept() error {
//请求接受数量和请求总量
accepts, total := b.history()
weightedAccepts := b.k * float64(accepts)
//计算丢弃请求概率
//https://landing.google.com/sre/sre-book/chapters/handling-overload/#eq2101
dropRatio := math.Max(0, (float64(total-protection)-weightedAccepts)/float64(total+1))
if dropRatio <= 0 {
return nil
}
//动态判断是否触发熔断
if b.proba.TrueOnProba(dropRatio) {
return ErrServiceUnavailable
}
return nil
}
func (b *googleBreaker) allow() (internalPromise, error) {
if err := b.accept(); err != nil {
return nil, err
}
return googlePromise{
b: b,
}, nil
}
func (b *googleBreaker) markSuccess() {
b.stat.Add(1)
}
func (b *googleBreaker) markFailure() {
b.stat.Add(0)
}
- 那么是怎么滑动计算的, 继续向下追查看markSuccess()或markFailure()方法中执行的Add(),默认情况下执行RollingWindow的Add()方法
参考博客
// Add adds value to current bucket.
func (rw *RollingWindow) Add(v float64) {
rw.lock.Lock()
defer rw.lock.Unlock()
//滑动的动作发生在此
rw.updateOffset()
rw.win.add(rw.offset, v)
}
// Reduce runs fn on all buckets, ignore current bucket if ignoreCurrent was set.
func (rw *RollingWindow) Reduce(fn func(b *Bucket)) {
rw.lock.RLock()
defer rw.lock.RUnlock()
var diff int
span := rw.span()
// ignore current bucket, because of partial data
if span == 0 && rw.ignoreCurrent {
diff = rw.size - 1
} else {
diff = rw.size - span
}
if diff > 0 {
offset := (rw.offset + span + 1) % rw.size
rw.win.reduce(offset, diff, fn)
}
}
func (rw *RollingWindow) span() int {
offset := int(timex.Since(rw.lastTime) / rw.interval)
if 0 <= offset && offset < rw.size {
return offset
}
return rw.size
}
func (rw *RollingWindow) updateOffset() {
span := rw.span()
if span <= 0 {
return
}
offset := rw.offset
//重置过期的 bucket
for i := 0; i < span; i++ {
rw.win.resetBucket((offset + i + 1) % rw.size)
}
rw.offset = (offset + span) % rw.size
now := timex.Now()
//更新时间
rw.lastTime = now - (now-rw.lastTime)%rw.interval
}
func (w *window) add(offset int, v float64) {
往执行的 bucket 加入指定的指标
w.buckets[offset%w.size].add(v)
}
服务端
- 在提供go-zero服务端时,执行zrpc.MustNewServer()创建RpcServer,内部最终会调用到一个NewServer()函数,该函数中:
- 执行c.HasEtcd()判断是否配置了etcd注册中心地址,如果配置了,执行NewRpcPubServer()函数
- 在NewRpcPubServer()中会创建一个名为registerEtcd的function函数,并将这个function封装到keepAliveServer结构体中
- 自此rpcServer创建成功,并封装了keepAliveServer结构体变量,内部持有一个注册服务的registerEtcd()函数,后续会通过这个函数是实现服务注册
- 执行setupInterceptors()函数注册拦截器
- 查看setupInterceptors()源码:这里要了解go-zero底层的keepAliveServer结构参考:go 进阶 go-zero相关: 四. 服务注册原理中提到
//setupInterceptors()根据配置信息为rpc服务添加一些拦截器
//入参： svr是一个rpc服务对象
// c是一个RpcServerConf结构体，包含了rpc服务的配置信息
// metrics是一个stat.Metrics对象，用于收集和报告统计指标
func setupInterceptors(svr internal.Server, c RpcServerConf, metrics *stat.Metrics) error {
//1.如果配置中指定了CpuThreshold参数，表示要开启自适应限流功能
if c.CpuThreshold > 0 {
// 创建一个自适应限流器对象，设置CPU阈值为配置中的值
shedder := load.NewAdaptiveShedder(load.WithCpuThreshold(c.CpuThreshold))
// 为rpc服务添加一个一元拦截器，用于执行限流逻辑，并记录统计指标
svr.AddUnaryInterceptors(serverinterceptors.UnarySheddingInterceptor(shedder, metrics))
}
//2.如果配置中指定了Timeout参数，表示要开启超时控制功能
if c.Timeout > 0 {
// 为rpc服务添加一个一元拦截器，用于执行超时控制逻辑，超时时间为配置中的值
svr.AddUnaryInterceptors(serverinterceptors.UnaryTimeoutInterceptor(
time.Duration(c.Timeout) * time.Millisecond))
}
//3.如果配置中指定了Auth参数，表示要开启鉴权功能
if c.Auth {
// 调用setupAuthInterceptors函数，为rpc服务添加鉴权相关的拦截器
if err := setupAuthInterceptors(svr, c); err != nil {
return err
}
}
return nil
}
- 继续查看用来添加拦截器的AddUnaryInterceptors()方法,最终会将拦截器保存到baseRpcServer的streamInterceptors属性,或unaryInterceptors属性中(下方源码在网上搜的,不知道是不是我的版本不对还是操作有问题在go-zero源码包中没找到…不能确认是否正确)
