go-zero 是如何实现令牌桶限流的？

// core/limit/tokenlimit.go
// 生成 token 速率script = `local rate = tonumber(ARGV[1])// 通容量local capacity = tonumber(ARGV[2])// 当前时间戳local now = tonumber(ARGV[3])// 请求数量local requested = tonumber(ARGV[4])// 需要多少秒才能把桶填满local fill_time = capacity/rate// 向下取整，ttl 为填满时间 2 倍local ttl = math.floor(fill_time*2)// 当前桶剩余容量，如果为 nil，说明第一次使用，赋值为桶最大容量local last_tokens = tonumber(redis.call("get", KEYS[1]))if last_tokens == nil then    last_tokens = capacityend
// 上次请求时间戳，如果为 nil 则赋值 0local last_refreshed = tonumber(redis.call("get", KEYS[2]))if last_refreshed == nil then    last_refreshed = 0end
// 距离上一次请求的时间跨度local delta = math.max(0, now-last_refreshed)// 距离上一次请求的时间跨度能生成的 token 数量和桶内剩余 token 数量的和// 与桶容量比较，取二者的小值local filled_tokens = math.min(capacity, last_tokens+(delta*rate))// 判断请求数量和桶内 token 数量的大小local allowed = filled_tokens >= requested// 被请求消耗掉之后，更新剩余 token 数量local new_tokens = filled_tokensif allowed then    new_tokens = filled_tokens - requestedend
// 更新 redis tokenredis.call("setex", KEYS[1], ttl, new_tokens)// 更新 redis 刷新时间redis.call("setex", KEYS[2], ttl, now)
return allowed`

// A TokenLimiter controls how frequently events are allowed to happen with in one second.type TokenLimiter struct {    // 生成 token 速率    rate           int    // 桶容量    burst          int    store          *redis.Redis    // 桶 key    tokenKey       string    // 桶刷新时间 key    timestampKey   string    rescueLock     sync.Mutex    // redis 健康标识    redisAlive     uint32    // redis 健康监控启动状态    monitorStarted bool    // 内置单机限流器    rescueLimiter  *xrate.Limiter}
// NewTokenLimiter returns a new TokenLimiter that allows events up to rate and permits// bursts of at most burst tokens.func NewTokenLimiter(rate, burst int, store *redis.Redis, key string) *TokenLimiter {    tokenKey := fmt.Sprintf(tokenFormat, key)    timestampKey := fmt.Sprintf(timestampFormat, key)
    return &TokenLimiter{        rate:          rate,        burst:         burst,        store:         store,        tokenKey:      tokenKey,        timestampKey:  timestampKey,        redisAlive:    1,        rescueLimiter: xrate.NewLimiter(xrate.Every(time.Second/time.Duration(rate)), burst),    }}

// Allow is shorthand for AllowN(time.Now(), 1).func (lim *TokenLimiter) Allow() bool {    return lim.AllowN(time.Now(), 1)}
// AllowCtx is shorthand for AllowNCtx(ctx,time.Now(), 1) with incoming context.func (lim *TokenLimiter) AllowCtx(ctx context.Context) bool {    return lim.AllowNCtx(ctx, time.Now(), 1)}
// AllowN reports whether n events may happen at time now.// Use this method if you intend to drop / skip events that exceed the rate.// Otherwise, use Reserve or Wait.func (lim *TokenLimiter) AllowN(now time.Time, n int) bool {    return lim.reserveN(context.Background(), now, n)}
// AllowNCtx reports whether n events may happen at time now with incoming context.// Use this method if you intend to drop / skip events that exceed the rate.// Otherwise, use Reserve or Wait.func (lim *TokenLimiter) AllowNCtx(ctx context.Context, now time.Time, n int) bool {    return lim.reserveN(ctx, now, n)}

func (lim *TokenLimiter) reserveN(ctx context.Context, now time.Time, n int) bool {    // 判断 Redis 健康状态，如果 Redis 故障，则使用进程内限流器    if atomic.LoadUint32(&lim.redisAlive) == 0 {        return lim.rescueLimiter.AllowN(now, n)    }
    // 执行限流脚本    resp, err := lim.store.EvalCtx(ctx,        script,        []string{            lim.tokenKey,            lim.timestampKey,        },        []string{            strconv.Itoa(lim.rate),            strconv.Itoa(lim.burst),            strconv.FormatInt(now.Unix(), 10),            strconv.Itoa(n),        })    // redis allowed == false    // Lua boolean false -> r Nil bulk reply    if err == redis.Nil {        return false    }    if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {        logx.Errorf("fail to use rate limiter: %s", err)        return false    }    if err != nil {        logx.Errorf("fail to use rate limiter: %s, use in-process limiter for rescue", err)        // 如果有异常的话，会启动进程内限流        lim.startMonitor()        return lim.rescueLimiter.AllowN(now, n)    }
    code, ok := resp.(int64)    if !ok {        logx.Errorf("fail to eval redis script: %v, use in-process limiter for rescue", resp)        lim.startMonitor()        return lim.rescueLimiter.AllowN(now, n)    }
    // redis allowed == true    // Lua boolean true -> r integer reply with value of 1    return code == 1}

func (lim *TokenLimiter) startMonitor() {    lim.rescueLock.Lock()    defer lim.rescueLock.Unlock()
    // 需要加锁保护，如果程序已经启动了，直接返回，不要重复启动    if lim.monitorStarted {        return    }
    lim.monitorStarted = true    atomic.StoreUint32(&lim.redisAlive, 0)
    go lim.waitForRedis()}
func (lim *TokenLimiter) waitForRedis() {    ticker := time.NewTicker(pingInterval)    // 更新监控进程的状态    defer func() {        ticker.Stop()        lim.rescueLock.Lock()        lim.monitorStarted = false        lim.rescueLock.Unlock()    }()
    for range ticker.C {        // 对 redis 进行健康监测，如果 redis 服务恢复了        // 则更新 redisAlive 标识，并退出 goroutine        if lim.store.Ping() {            atomic.StoreUint32(&lim.redisAlive, 1)            return        }    }}