Redis限流

【参考】

《Redis深度历险 核心原理与应用实践》
https://blog.csdn.net/liubao616311/article/details/89247622

【引】

1.当系统处理能力有限时，如何进行限流？
2.控制用户行为，避免垃圾请求（避免短时间多次操作）。比如一段时间内用户只能操作一个接口1次，或者一段时间内用户只能操作某个行为N次？
【一些简单方案的讨论】
之前的做法是用interface_name:name组成的key，设置为N秒过期，那么这个用户如果没有这个key，那么可以进行设置并执行调用接口，但如果已经存在，返回操作频繁。
对于接口一段时间只能操作1次，我们可以这么做，不仅简单而且还能避免一些并发的问题。这里把场景改一下：
假设我们为了防止用户恶意操作带来的服务资源占用，给用户设定了1分钟内最多操作某个接口3次，又该如何处理呢？
这里和规定时间1次的限制不同的是，我们必须从当前时间往前推1分钟，这个时间窗口是动态的，也就没办法用固定的过期时间来判断。比较容易想到的做法就是去数据库存下这个用户上次访问的记录以及时间，当用户这一次访问来了之后，先找到时间窗口，然后到数据库找这段时间limit 3个历史记录，如果找到了说明这一次操作不能执行。但是这样就会造成大量的数据库查询，当用户较多或者这样的操作比较频繁时，会对系统造成较大的压力。
【Redis中的简单限流】
先来复习一下zset结构的几个命令含义：
zadd表示添加元素，zremrangebyscore表示将范围内的元素剔除，zcard表示统计结构中的元素个数，zrange表示获取范围内的元素，如下：

针对上面的1分钟3次限制的场景，Redis的思路是用zset结构存储这种带时间窗口的结构：
1)zadd key timestamp String(timestamp) #存入当前的时间
2)zremrangeByScore key 0 timestamp-period*1000 #剔除掉0到当前时间-限制时间这一段的所有值，即把有效期内的数据都剔除掉
3)zcard key #统计此时的有效元素数量
4)expire key period+1 #设置过期时间，留下余量
然后将这四个指令用管道的形式进行执行，减少RTT(Round-trip time)。根据这组命令可以捋一下思路：
将此次请求加入，然后计算有效时间内的次数，如果次数大于等于（本次已算入内）限定次数，那么返回false，否则就执行用户这次操作。
但这里也有一个问题，返回false的时候，这一次是不会执行的，但是还是记录到了这个zset中。所以想要严格记录已经执行成功的次数并以此为依据进行数据的筛选，可以将234步骤的执行先于1执行，得到结果是true，再将本次的请求zadd进入，这里需要注意的是这两个操作之间如何保证原子性，不然在zadd之前，真正的统计结果已经发生变化，就会出现判断失误。或者我们也可以使用lua脚本进行这个逻辑的判断，这样就节省了RTT，也保证了原子性。
【漏斗限流原理】
再想一个问题，比如一个消息队列，它的容量有限，假设队列的消费者消费速率一定，那么我们在向消息队列存入元素的时候如何判断本次存入的元素数量是否会将队列容量爆仓呢（有点像小学的追击问题...）？
这里有一个比较有名的漏斗限流算法。上面的问题其实可以看成服务器处理请求的能力以及现有资源来判断新来的请求是否要去执行，每次执行操作之前，根据服务器的处理速度和上一次请求打入的时间进行计算是否可以接受本次请求的处理，看JAVA代码：

package com.redis.throttle;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class FunnelRateLimiter {

    static class Funnel {

        // 漏斗容量
        int capacity;
        // 漏嘴流水速率
        float leakingRate;
        // 漏斗剩余空间
        int leftQuota;
        // 上一次漏水时间
        long leakingTs;

        public Funnel(int capacity, float leakingRate) {
            this.capacity = capacity;
            this.leakingRate = leakingRate;
            this.leftQuota = capacity;
            this.leakingTs = System.currentTimeMillis();
        }

        void makeSpace() {
            // 当前时间
            long nowTs = System.currentTimeMillis();
            // 距离上一次漏水过去了多长时间
            long deltaTs = nowTs - leakingTs;
            // 过去的时间流了多少水
            int deltaQuota = (int) (deltaTs * leakingRate);
            // 小于0说明溢出了，当前漏斗已经是空的了
            if (deltaQuota < 0) {
                this.leakingRate = capacity;
                this.leakingTs = nowTs;
                return;
            }
            // 流出的还没到1，不用计算了
            if (deltaQuota < 1) {
                return;
            }
            // 剩余空间大于0，但是和流出的相加超出最大值后，也重新初始化容量
            if (this.leftQuota > 0 && (this.leftQuota + deltaQuota) < 0) {
                this.leakingRate = capacity;
                this.leakingTs = nowTs;
                return;
            }
            // 将剩余容量和流出的累加
            this.leftQuota += deltaQuota;
            // 记录本次流水时间
            this.leakingTs = nowTs;
            // 超出最大值，则用最大值
            if (this.leftQuota > this.capacity) {
                this.leftQuota = this.capacity;
            }
        }

        boolean watering(int quota) {
            // 计算流水和剩余容量
            makeSpace();
            // 剩余容量大于等于本次的用量，允许操作
            if (this.leftQuota >= quota) {
                this.leftQuota -= quota;
                return true;
            }
            return false;
        }
    }

    private static Map<String, Funnel> funnelMap = new ConcurrentHashMap<>();

    public static boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {
        String key = String.format("%s:%s", userId, actionKey);
        Funnel funnel = funnelMap.get(key);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate);
            funnelMap.put(key, funnel);
        }
        return funnel.watering(1);
    }

    public static void main(String[] args) {
        for (int i = 0; i < 20; i++) {
            System.out.println(isActionAllowed("mn", "reply", 5, 1));
        }
    }

}

对于单机这样的计算是没问题的，但是对于分布式系统来说，多个客户端同时操作时，我们没办法保证这些操作是原子性的（从map中获取到key，逻辑判断，并将新值存入map中）。所以必须要加入分布式锁进行控制，但加锁就意味着抢锁失败（可能此时还有足够的空间进行处理，只是因为没有抢到锁导致操作失败），在抢锁失败后性能（重试）和用户体验（放弃本次操作）我们只能二选一。
【Redis的cl.throttle】
throttle意为限流阀，Rdis4.0提供了一个Redis-Cell模块，采用漏斗算法，并提供了原子指令。有这个模块，可以方便我们进行限流操作。
如图展示了cl.throttle的参数：

--安装教程--
wget https://github.com/brandur/redis-cell/releases/download/v0.3.0/redis-cell-v0.3.0-x86_64-unknown-linux-gnu.tar.gz
# 下载不下来也可以按照大目录的地址去网页上直接下载
tar -vxfz redis-cell-v0.3.0-x86_64-unknown-linux-gnu.tar.gz

接下来到redis-conf中loadmodule，再restart服务即可
--Java测试代码--

public enum CellCommand implements ProtocolCommand {
    CLTHROTTLE("CL.THROTTLE");

    private final byte[] raw;
    CellCommand(String alt) {
        raw = SafeEncoder.encode(alt);
    }
    public byte[] getRaw() {
        return raw;
    }
}

public class ThrottleTest {
    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = new Jedis("localhost", 6379);
        jedis.auth("123456");
        for (int i = 0; i < 50; i++) {
            Connection client = jedis.getClient();
            client.sendCommand(CellCommand.CLTHROTTLE, "user123", "15", "30", "60", "2");
            List<Long> replay = client.getIntegerMultiBulkReply();
            if (replay.get(0) == 0) {
                System.out.println("系统可用，" + i + "号用户可以通过");
                Thread.sleep(1000);
            } else {
                System.out.println("系统繁忙，" + i + "号用户限制通过，请稍候重试！");
            }
            client.close();
        }
    }
}

最后的结果：