redis限流方法 | 小码农博客

1. 利用之前的redis锁来处理队列请求

利用set nx 原子操作强行排队

2. 使用zset有序集合限定某一时间内的流量

原理

zset为有序集合, 记录下某一key值,相同分数, 不同访问时间
zadd user:10 0 1617070146897, 请求到来时, 获取到(限流总数-1)的有序数据量, 即当前限制key的次数访问记录, 判断第一个值的时间, 是否为在限制区间内

相关代码:

class RateLimiter {
    /**
     * 
     * @param {String} type 限制类型, user|uri|ip, 可组合, user,uri|uri,ip|user,ip|user,uri,ip
     */
    constructor(type, msg) {
        // 用于判断生成 zset 的 key
        // 空,则仅对user进行判断
        this.types = (type || 'user').split(',');
        this.msg = msg;
        this.rules = [];
    }

    getKey(req) {
        let keys = [];
        for(let item of this.types) {
            switch(item) {
                case 'user':
                    key.push(`user:${req.userIfo.id}`)
                    break;
                case 'uri':
                    key.push(`uri:${req.url}`)
                    break;
                case 'ip':
                    key.push(`ip:${req.get('X-Real-IP') || req.ip}`)
                    break;
            }
        }
        return keys.join(':');
    }

    rate(rules) {
        if (!rules) throw new Error("ReteLimiter.rate is not rules");
        this.rules = (rules || []).sort((a,b) => a.time > b.time);
        return this.action
    }

    action(req, res, next) {
        let key = this.getKey(req);
        // 组织定义对象
        let redis = req.redis;
        for(var item of this.rules) {
           let limit = await this.query(redis, key, item);
           if (limit) {
               res.statusCoce(429);
               res.send({"msg": this.msg || 'Many too request'})
               return res.end();
           }
        }

        await next();
        // 记录
        this.mark(redis, key, this.rules[this.rules.length-1].time);
    }

    async mark(redis, key, maxRule) {
        let time = new Date().getTime();
        redis.zadd(key, time, time);
        // 移除过期
        let remTime = time - maxRule * 1000 - 1;
        let count = await redis.zcount(key, 0, remTime);
        if (count>0) {
            await redis.zremrangebysore(key, 0, remTime);
        }
    }

    /**
     * 
     * @param {RedisConection} redis 
     * @param {Object} ops 查询单位
     * @param {Number} ops.time 单位时长, 单位:s, 1s, 300s, 3600s等
     * @param {Number} ops.rate 单位时间限制次数 
     * @returns {Boolean} 是否被限制 
     */
    async query(redis, key, {time, rate}) {
        let now = new Date().getTime();
        let result = await redis.zrange(key, -1 * rate, -1 * rate);
        if (now - result[0] < time * 1000) return true;
        return false
    }
}

3. 使用令牌桶算法限流

漏桶
以一定速度生成令牌, 放入桶内, 请求来临时, 桶内以固定速率释放令牌, 请求获取到令牌,则继续进行, 无峰值访问, 实际请求强制排队等候, 类似setnx, 可以用 lpush/blpop模拟实现, 假定生成与释放令牌速度相同, 即定时生成即消费

漏桶

# 生成
lpush user:10:$url $token
# 获取
blpop user:10:$url

令牌桶
以一定速度生成令牌, 放入桶内, 请求来临时,从桶内获取令牌, 如果桶内为空则拒绝请求, 当请求速度大于令牌生成速度时, 桶内在一定时间内为空, 产生拒绝服务, 可以产生较小峰值

共同缺点, 需要额外进程/线程以固定速率生成令牌放入队列, 利用的是 lpush/blpop/llen适合通用接口流量削峰填谷

令牌桶

# 额外进程生成令牌
llen user:10:$url => 如果大于设定值如100, 则不再继续生成
lpush user:10:$url $token

# 请求进入(redis blpop先到先得,不会超发)
blpop user:10:$url 
# 拿到令牌继续操作