Ein Artikel, um über die aktuelle Begrenzungsstrategie in Redis zu sprechen

In diesem Artikel erfahren Sie mehr über die Strombegrenzung in Redis und stellen die einfache Strombegrenzungsstrategie und die Trichterstrombegrenzung vor. Ich hoffe, dass er für alle hilfreich ist!

1. Einfache Strombegrenzung

Grundprinzipien

Wenn die Systemverarbeitungskapazität begrenzt ist, wie können ungeplante Anforderungen organisiert werden, um Druck auf das System auszuüben? Schauen wir uns zunächst einige einfache Strategien zur Strombegrenzung an, um Brute-Force-Angriffe zu verhindern. Wenn Sie beispielsweise auf eine IP zugreifen möchten, können Sie alle 5 Sekunden nur 10 Mal darauf zugreifen. Wenn das Limit überschritten wird, wird sie blockiert. [Verwandte Empfehlungen: Redis-Video-Tutorial]

Wie oben gezeigt, wird im Allgemeinen ein Schiebefenster verwendet, um die Anzahl der Besuche innerhalb eines Intervalls zu zählen. Verwenden Sie zset, um die Anzahl der IP-Besuche aufzuzeichnen. Jede IP wird über key und score Speichern Sie den aktuellen Zeitstempel, <code>value wird nur mit Zeitstempel oder UUID implementiertzset 记录 IP 访问次数，每个 IP 通过 key 保存下来，score 保存当前时间戳，value 唯一用时间戳或者UUID来实现

代码实现

public class RedisLimiterTest {
    private Jedis jedis;

    public RedisLimiterTest(Jedis jedis) {
        this.jedis = jedis;
    }

    /**
     * @param ipAddress Ip地址
     * @param period    特定的时间内，单位秒
     * @param maxCount  最大允许的次数
     * @return
     */
    public boolean isIpLimit(String ipAddress, int period, int maxCount) {
        String key = String.format("ip:%s", ipAddress);
        // 毫秒时间戳
        long currentTimeMillis = System.currentTimeMillis();
        Pipeline pipe = jedis.pipelined();
        // redis事务，保证原子性
        pipe.multi();
        // 存放数据，value 和 score 都使用毫秒时间戳
        pipe.zadd(key, currentTimeMillis, "" + UUID.randomUUID());
        // 移除窗口区间所有的元素
        pipe.zremrangeByScore(key, 0, currentTimeMillis - period * 1000);
        // 获取时间窗口内的行为数量
        Response<Long> count = pipe.zcard(key);
        // 设置 zset 过期时间，避免冷用户持续占用内存，这里宽限1s
        pipe.expire(key, period + 1);
        // 提交事务
        pipe.exec();
        pipe.close();
        // 比较数量是否超标
        return count.get() > maxCount;
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379);
        RedisLimiterTest limiter = new RedisLimiterTest(jedis);
        for (int i = 1; i <= 20; i++) {
            // 验证IP  10秒钟之内只能访问5次
            boolean isLimit = limiter.isIpLimit("222.73.55.22", 10, 5);
            System.out.println("访问第" + i + "次, 结果：" + (isLimit ? "限制访问" : "允许访问"));
        }
    }
}

执行结果

访问第1次, 结果：允许访问
访问第2次, 结果：允许访问
访问第3次, 结果：允许访问
访问第4次, 结果：允许访问
访问第5次, 结果：允许访问
访问第6次, 结果：限制访问
访问第7次, 结果：限制访问
... ...

缺点：要记录时间窗口所有的行为记录，量很大，比如，限定60s内不能超过100万次这种场景，不太适合这样限流，因为会消耗大量的储存空间。

二、漏斗限流

基本原理

• 漏斗的容量是限定的，如果满了，就装不进去了。
• 如果将漏嘴放开，水就会往下流，流走一部分之后，就又可以继续往里面灌水。
• 如果漏嘴流水的速率大于灌水的速率，那么漏斗永远都装不满。
• 如果漏嘴流水速率小于灌水的速率，那么一旦漏斗满了，灌水就需要暂停并等待漏斗腾空。

示例代码

public class FunnelLimiterTest {

    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); // 腾出的空间 = 时间*漏水速率
            if (deltaQuota < 0) { // 间隔时间太长，整数数字过大溢出
                this.leftQuota = capacity;
                this.leakingTs = nowTs;
                return;
            }
            if (deltaQuota < 1) { // 腾出空间太小 就等下次，最小单位是1
                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 Map<String, Funnel> funnels = new HashMap<>();

    /**
     * @param capacity    漏斗容量
     * @param leakingRate 漏嘴流水速率 quota/s
     */
    public boolean isIpLimit(String ipAddress, int capacity, float leakingRate) {
        String key = String.format("ip:%s", ipAddress);
        Funnel funnel = funnels.get(key);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate);
            funnels.put(key, funnel);
        }
        return !funnel.watering(1); // 需要1个quota
    }

    public static void main(String[] args) throws Exception{
        FunnelLimiterTest limiter = new FunnelLimiterTest();
        for (int i = 1; i <= 50; i++) {
            // 每1s执行一次
            Thread.sleep(1000);
            // 漏斗容量是2 ，漏嘴流水速率是0.5每秒，
            boolean isLimit = limiter.isIpLimit("222.73.55.22", 2, (float)0.5/1000);
            System.out.println("访问第" + i + "次, 结果：" + (isLimit ? "限制访问" : "允许访问"));
        }
    }
}

执行结果

访问第1次, 结果：允许访问    # 第1次，容量剩余2，执行后1
访问第2次, 结果：允许访问    # 第2次，容量剩余1，执行后0
访问第3次, 结果：允许访问    # 第3次，由于过了2s, 漏斗流水剩余1个空间，所以容量剩余1，执行后0
访问第4次, 结果：限制访问    # 第4次，过了1s, 剩余空间小于1， 容量剩余0
访问第5次, 结果：允许访问    # 第5次，由于过了2s, 漏斗流水剩余1个空间，所以容量剩余1，执行后0
访问第6次, 结果：限制访问    # 以此类推...
访问第7次, 结果：允许访问
访问第8次, 结果：限制访问
访问第9次, 结果：允许访问
访问第10次, 结果：限制访问

• 我们观察 Funnel 对象的几个字段，我们发现可以将 Funnel 对象的内容按字段存储到一个 hash 结构中，灌水的时候将 hash 结构的字段取出来进行逻辑运算后，再将新值回填到 hash 结构中就完成了一次行为频度的检测。
• 但是有个问题，我们无法保证整个过程的原子性。从 hash 结构中取值，然后在内存里运算，再回填到 hash 结构，这三个过程无法原子化，意味着需要进行适当的加锁控制。而一旦加锁，就意味着会有加锁失败，加锁失败就需要选择重试或者放弃。
• 如果重试的话，就会导致性能下降。如果放弃的话，就会影响用户体验。同时，代码的复杂度也跟着升高很多。这真是个艰难的选择，我们该如何解决这个问题呢？Redis-Cell 救星来了！

Redis-Cell

Redis 4.0 提供了一个限流 Redis 模块，它叫 redis-cell。该模块也使用了漏斗算法，并提供了原子的限流指令。 该模块只有1条指令cl.throttle，它的参数和返回值都略显复杂，接下来让我们来看看这个指令具体该如何使用。

> cl.throttle key:xxx 15 30 60 1

• 15 : 15 capacity 这是漏斗容量
• 30 60 : 30 operations / 60 seconds 这是漏水速率
• 1 : need 1 quota (可选参数，默认值也是1)

> cl.throttle laoqian:reply 15 30 60
1) (integer) 0   # 0 表示允许，1表示拒绝
2) (integer) 15  # 漏斗容量capacity
3) (integer) 14  # 漏斗剩余空间left_quota
4) (integer) -1  # 如果拒绝了，需要多长时间后再试(漏斗有空间了，单位秒)
5) (integer) 2   # 多长时间后，漏斗完全空出来(left_quota==capacity，单位秒)

在执行限流指令时，如果被拒绝了，就需要丢弃或重试。cl.throttle 指令考虑的非常周到，连重试时间都帮你算好了，直接取返回结果数组的第四个值进行 sleep

Code-Implementierung

rrreee

Ausführungsergebnisse🎜🎜rrreee🎜Nachteile: Es ist notwendig, alle Verhaltensdatensätze im Zeitfenster aufzuzeichnen, was beispielsweise sehr groß ist. Das Szenario, die Häufigkeit auf nicht mehr als 1 Million Mal in 60 Sekunden zu begrenzen, ist dafür nicht geeignet Strombegrenzung, da dadurch viel Speicherplatz verbraucht wird. 🎜

2. Strombegrenzung des Trichters 🎜

🎜Grundprinzip🎜🎜

• Die Kapazität des Der Trichter ist begrenzt. Wenn er voll ist, passt er nicht hinein.
• Wenn Sie den Auslauf loslassen, fließt das Wasser nach unten. Nachdem ein Teil davon abfließt, können Sie ihn weiter mit Wasser füllen.
• Wenn die Durchflussrate des Wassers aus dem Auslauf größer ist als die Wasserfüllrate, wird der Trichter nie voll sein.
• Wenn die Durchflussrate des Trichters geringer ist als die Füllrate, muss der Füllvorgang angehalten werden, sobald der Trichter voll ist, und warten, bis der Trichter leer ist.

🎜Beispielcode🎜🎜rrreee🎜🎜Ausführungsergebnisse🎜🎜rrreee

• Wir beobachten den Trichter Bei mehreren Feldern haben wir festgestellt, dass der Inhalt des Funnel-Objekts feldweise in einer hash-Struktur gespeichert werden kann Nachdem die Felder für logische Operationen entnommen wurden, werden die neuen Werte in die hash-Struktur eingefügt, um eine Verhaltenshäufigkeitserkennung abzuschließen.
• Aber es gibt ein Problem: Wir können die Atomizität des gesamten Prozesses nicht garantieren. Holen Sie sich den Wert aus der hash-Struktur, verarbeiten Sie ihn dann im Speicher und füllen Sie ihn dann in die hash-Struktur ein. Diese drei Prozesse können nicht atomar sein, was bedeutet, dass eine entsprechende Sperre erforderlich ist Kontrolle ist erforderlich. Sobald die Sperre gesperrt ist, liegt ein Sperrfehler vor. Wenn die Sperre fehlschlägt, müssen Sie entscheiden, ob Sie es erneut versuchen oder aufgeben möchten.
• Wenn Sie es erneut versuchen, verringert sich die Leistung. Wenn Sie aufgeben, wirkt sich dies auf die Benutzererfahrung aus. Gleichzeitig hat auch die Komplexität des Codes stark zugenommen. Das ist eine wirklich schwierige Entscheidung. Wie lösen wir dieses Problem? Redis-Cell Der Retter ist da!

🎜Redis-Cell🎜🎜🎜Redis 4.0 bietet ein aktuell begrenztes Redis-Modul namens redis-cell. Dieses Modul verwendet auch den Trichteralgorithmus und bietet Anweisungen zur atomaren Strombegrenzung. Dieses Modul hat nur eine Anweisung cl.throttle und seine Parameter und Rückgabewerte sind etwas kompliziert. Schauen wir uns als Nächstes an, wie diese Anweisung verwendet wird. 🎜rrreee

• 15 : 15 Kapazität Das ist die Trichterkapazität
• 30 60 : 30 Betätigungen / 60 Sekunden Das ist das Wasser Leckrate
• 1 : benötigt 1 Kontingent (optionaler Parameter, der Standardwert ist ebenfalls 1)

rrreee🎜Wenn der Strombegrenzungsbefehl ausgeführt wird, wenn Es wird abgelehnt. Sie müssen es verwerfen oder es erneut versuchen. Die Anweisung cl.throttle ist sehr durchdacht und berechnet sogar die Wiederholungszeit für Sie. Nehmen Sie einfach den vierten Wert des zurückgegebenen Ergebnisarrays und führen Sie sleep aus. Wenn Sie den Thread nicht blockieren möchten, können Sie es auch mit einer asynchronen geplanten Aufgabe erneut versuchen. 🎜🎜Weitere Kenntnisse zum Thema Programmierung finden Sie unter: 🎜Programmiervideos🎜! ! 🎜

Das obige ist der detaillierte Inhalt vonEin Artikel, um über die aktuelle Begrenzungsstrategie in Redis zu sprechen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!