限流

当系统的处理能力有限时，如何阻止计划外的请求继续对系统施压，是一个需要重视的问题，避免超出负载的流量影响系统的稳定运行，这就需要用到限流算法，

除了控制流量，限流还有一个目的是控制用户行为，避免垃圾请求，比如在论坛上，用户的发帖、回复、点赞等行为都要严格受控，一般短时间内用户的请求将会收到一定次数的限制，超过这个限制将拒绝或做其它处理。

使用redis简单限流

接下来我们使用redis实现一个在指定时间内只能做固定次数的操作，超出这个次数，就做出拒绝处理，

我们可以考虑使用zset这个数据结构，使用score存储每次操作的时间戳，value根据业务情况来，保证value唯一性即可，

随后每次我们使用滑动时间窗口(定宽)，每次都保留着这个窗口内的数据，其余的都trim掉，此举也可避免一定的内存空间的浪费，如果用户是冷用户，滑动时间窗口内的行为是空记录，那么该zset就可以从内存中移除，不再占用空间，

使用zset结构记录用户的行为历史，每一个行为都会作为zset中的一个key保存下来，同一个用户的同一种行为用一个zset记录，

下方使用zset的滑动窗口代码实现，我们首先在zset中添加用户本次行为，然后删除了本次时间窗口外的数据，

最后获取本次窗口内的数据总数，如果总数没有超出限制，我们返回true，超出限制了返回false，下面main方法中设置的时间窗口是60s，在60s內最多5次請求，

根据结果我们可以看到，循环20次，只有前5次得到了true，允许请求访问，后续的结果全是false，被拒绝访问。

因为这几个连续的操作都是针对同一个key的，使用pipeline可以显著提高redis存取效率。

不过这种方案存在着自己的不足，因为要记录时间窗口内所有的行为记录，如果这个量很大，比如60s内操作不能超过100万次，这时候是不适合用这种方式的，会消耗大量的存储空间。

漏斗限流

漏斗限流是最常用的限流方法之一，这个算法的灵感来自于漏斗(funnel)的结构，如下图所示，漏斗的容量是有限的，

如果将漏嘴堵住，然后一直灌水，它会变满，直到再也装不下，如果将漏嘴放开，水会从下方流出，流走一部分后，又可以继续灌水，

如果漏嘴的速率大于灌水的速率，那么漏洞永远装不满，如果漏嘴速率小于灌水的速率，一旦漏斗满了，灌水就需要暂停并等待漏斗腾出一部分的空间，方可继续灌水,

所以，漏斗的剩余空间就代表着当前行为可以持续进行的数量，漏嘴的流水速率代表着系统允许该行为的最大频率。

java实现

package com.redis.cell;

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();
        }

        synchronized 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;
            }
        }

        synchronized 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));
        }
    }

}

Funnel对象的makeSpace方法是漏斗算法的核心，每次灌水前都会调用该方法触发漏水，给漏斗腾出空间，能腾出多少空间取决过去了多久，以及流水的速率，

Funnel对象占据的空间大小不再和行为的频率成正比，它的空间占用是一个常量。

上述代码的运行结果如下图，速率是1毫秒流走一个量，最大容量是5，可以看到下面的输出，输出5个true后，后续的输出都是false，直到过去1毫秒后，才继续输出true。

上面的代码是使用Java代码实现的，实现分布式则需要使用redis来处理，我们可以使用redis的hash结构，

将Funnel对象的字段存储到hash中去，灌水的时候将hash中的字段取出运算，最后再放回去，但是这就出现了原子性的问题，因为这几个步骤的过程非原子性，意味着需要进行加锁，加锁还有加锁失败重试的可能，降低了性能，影响用户体验，

不过这些操作可以使用lua脚本来做，但是在redis4.0中为我们提供了一个Redis-Cell的模块，这个模块使用了漏斗算法实现了限流，并提供原子的限流指令给我们用，这使得用redis实现限流就很简单了。

RedisCell

该模块需要额外安装插件，我们使用docker进行安装运行:

docker pull carto/redis-cell
docker run -d -p 6379:6379 --name redisCell 69846d418101

该模块只有1个指令，cl.throttle，它的参数和返回值都比较多，下面看看使用说明

上面命令的意思是，允许用户点赞的行为频率为60s内共30次操作，漏斗的初始容量为15，即开始连续可以点15次的赞，后续将受到漏水速率的影响，

所以上面的命令中 user1:zan 是用户的id+行为(key)，15是容量(capacity)，30是共多少次操作(operations)，60则是多长时间(seconds)，最后个即每次操作消耗的容量(quota)，可以不填，默认也是1。

返回的结果解释如下

127.0.0.1:6379> cl.throttle user1:zan 15 30 60 1

1) (integer) 0   # 0 表示允许，1表示拒绝

2) (integer) 16  # 漏斗容量

3) (integer) 15  # 漏洞剩余空间

4) (integer) -1  # 被拒绝后，还有多少秒漏洞会有空间（多少秒后可以重试）

5) (integer) 2   # 多长时间后，漏斗将完全空出来（单位秒）

执行限流命令时，如果被拒绝了，就需要丢弃或重试，cl.throttle 指令将多久后重试的时间都算好了，直接取返回结果的第四个值进行sleep等待后重试即可，如果不想阻塞线程，也可以用异步定时任务重试。

Java代码运行结果如下：

本文用到的代码都在：https://github.com/qiaomengnan16/redis-demo/tree/main/redis-cell