漏斗限流详述

Posted 李子捌

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了漏斗限流详述相关的知识,希望对你有一定的参考价值。

本文已收录于专栏

❤️《Redis之大厂必备技能包》❤️

上千人点赞收藏的,全套Redis学习资料,大厂必备技能!


目录

1、需求

2、常见的错误设计

3、漏斗限流

3.1 解决方案

3.2 Java代码实现

3.3 结合Redis实现

4、总结


1、需求

限定用户的某个行为在指定时间T内,只允许发生N次。假设T为1秒钟,N为1000次。

2、常见的错误设计

程序员设计了一个在每分钟内只允许访问1000次的限流方案,如下图01:00s-02:00s之间只允许访问1000次,这种设计最大的问题在于,请求可能在01:59s-02:00s之间被请求1000次,02:00s-02:01s之间被请求了1000次,这种情况下01:59s-02:01s间隔0.02s之间被请求2000次,很显然这种设计是错误的。

3、漏斗限流

3.1 解决方案

漏斗容量有限,当流水的的速度小于灌水的速度,漏斗就会水满溢出,利用这个原理我们可以设计限流代码!漏斗的剩余的空间就代表着当前行为(请求)可以持续进行的数量,漏斗的流水速率代表系统允许行为(请求)发生的最大频率,通常安装系统的处理能力权衡后进行设值。

3.2 Java代码实现

package com.lizba.redis.limit;

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

/**
 * <p>
 *      漏斗限流
 * </p>
 *
 * @Author: Liziba
 */
public class FunnelRateLimiter {

    /** map用于存储多个漏斗 */
    private Map<String, Funnel> funnels = new ConcurrentHashMap<>();

    /**
     * 请求(行为)是否被允许
     *
     * @param userId        用户id
     * @param actionKey     行为key
     * @param capacity      漏斗容量
     * @param leakingRate   剩余容量
     * @param quota         请求次数
     * @return
     */
    public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate, int quota) {

        String key = String.format("%s:%s", userId, actionKey);
        Funnel funnel = funnels.get(key);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate);
            funnels.put(key, funnel);
        }
        return funnel.waterLeaking(quota);
    }

    /**
     * 漏斗类
     */
    class Funnel {
        /** 漏斗容量 */
        int capacity;
        /** 漏斗流速,每毫秒允许的流速(请求) */
        float leakingRate;
        /** 漏斗剩余空间 */
        int leftCapacity;
        /** 上次漏水时间 */
        long leakingTs;

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

        /**
         * 计算剩余空间
         */
        void makeSpace() {
            long nowTs = System.currentTimeMillis();
            long intervalTs = nowTs - leakingTs;
            int intervalCapacity = (int) (intervalTs * leakingRate);
            // int 溢出
            if (intervalCapacity < 0) {
                this.leftCapacity = this.capacity;
                this.leakingTs = nowTs;
                return;
            }
            // 腾出空间 >= 1
            if (intervalCapacity < 1) {
                return;
            }
            // 增加漏斗剩余容量
            this.leftCapacity += intervalCapacity;
            this.leakingTs = nowTs;
            // 容量不允许超出漏斗容量
            if (this.leftCapacity > this.capacity) {
                this.leftCapacity = this.capacity;
            }
        }

        /**
         * 漏斗流水
         *
         * @param quota     流水量
         * @return
         */
        boolean waterLeaking(int quota) {
            // 触发漏斗流水
            this.makeSpace();
            if (this.leftCapacity >= quota) {
                leftCapacity -= quota;
                return true;
            }
            return false;
        }

    }

}

测试代码:
计算机运行如下的代码速度会非常的块,我通过TimeUnit.SECONDS.sleep(2);模拟客户端过一段时间后再请求。
设置漏斗容量为10,每毫秒允许0.002次请求(2 次/秒),每次请求数量为1;

package com.lizba.redis.limit;

import java.util.concurrent.TimeUnit;

/**
 * @Author: Liziba
 */
public class TestFunnelRateLimit {

    public static void main(String[] args) throws InterruptedException {

        FunnelRateLimiter limiter = new FunnelRateLimiter();
        for (int i = 1; i <= 20; i++) {
            if (i == 15) {
                TimeUnit.SECONDS.sleep(2);
            }
            // 设置漏斗容量为10,每毫秒允许0.002次请求(2 次/秒),每次请求数量为1
            boolean success = limiter.isActionAllowed("Liziba", "commit", 10, 0.002f, 1);
            System.out.println("第" + i + "请求" + (success ? "成功" : "失败"));
        }

    }

}

测试结果:

  1. 01-10次请求成功,初始漏斗大小为10,因此前10次请求成功
  2. 11-14次请求失败,由于漏斗已满,并且漏斗的流速在这四次请求之间未能释放1
  3. 15-18次请求成功,因为i == 15时主线程睡眠2秒,2秒时间漏斗流出 0.00210002 = 4,因此这四次请求成功
  4. 19-20次请求失败,与11-14次请求失败的原因一致

3.3 结合Redis实现

我们采用hash结构,将Funnel的属性字段,放入hash中,并且在代码中进行运算即可

package com.lizba.redis.limit;

import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Map;

/**
 * <p>
 *      redis hash 漏斗限流
 * </p>
 *
 * @Author: Liziba
 * @Date: 2021/9/7 23:46
 */
public class FunnelRateLimiterByHash {

    private Jedis client;

    public FunnelRateLimiterByHash(Jedis client) {
        this.client = client;
    }

    /**
     * 请求是否成功
     *
     * @param userId
     * @param actionKey
     * @param capacity
     * @param leakingRate
     * @param quota
     * @return
     */
    public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate, int quota) {
        String key = this.key(userId, actionKey);
        long nowTs = System.currentTimeMillis();
        Map<String, String> funnelMap = client.hgetAll(key);
        if (funnelMap == null || funnelMap.isEmpty()) {
            return initFunnel(key, nowTs, capacity, quota);
        }
        long intervalTs = nowTs - Long.parseLong(funnelMap.get("leakingTs"));
        int intervalCapacity = (int) (intervalTs * leakingRate);
        // 时间过长, int可能溢出
        if (intervalCapacity < 0) {
            intervalCapacity = 0;
            initFunnel(key, nowTs, capacity, quota);
        }
        // 腾出空间必须 >= 1
        if (intervalCapacity < 1) {
            intervalCapacity = 0;
        }
        int leftCapacity = Integer.parseInt(funnelMap.get("leftCapacity")) + intervalCapacity;
        if (leftCapacity > capacity) {
            leftCapacity = capacity;
        }
        return initFunnel(key, nowTs, leftCapacity, quota);
    }



    /**
     * 存入redis,初始funnel
     *
     * @param key
     * @param nowTs
     * @param capacity
     * @param quota
     * @return
     */
    private boolean initFunnel(String key,long nowTs, int capacity, int quota) {
        Map<String, String> funnelMap = new HashMap<>();
        funnelMap.put("leftCapacity", String.valueOf((capacity > quota) ? (capacity - quota) : 0));
        funnelMap.put("leakingTs", String.valueOf(nowTs));
        client.hset(key, funnelMap);
        return capacity >= quota;
    }

    /**
     * 限流key
     *
     * @param userId
     * @param actionKey
     * @return
     */
    private String key(String userId, String actionKey) {
        return String.format("limit:%s:%s", userId, actionKey);
    }

}

测试代码:

package com.lizba.redis.limit;

import redis.clients.jedis.Jedis;

import java.util.concurrent.TimeUnit;

/**
 * @Author: Liziba
 */
public class TestFunnelRateLimiterByHash {

    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = new Jedis("192.168.211.108", 6379);

        FunnelRateLimiterByHash limiter = new FunnelRateLimiterByHash(jedis);
        for (int i = 1; i <= 20; i++) {
            if (i == 15) {
                TimeUnit.SECONDS.sleep(2);
            }
            boolean success = limiter.isActionAllowed("liziba", "view", 10, 0.002f, 1);
            System.out.println("第" + i + "请求" + (success ? "成功" : "失败"));
        }

        jedis.close();
    }

}

测试结果:
与上面的java代码结构一致

4、总结

上述说了两种实现漏斗限流的方式,其实思想都是一样的,但是这两者都无法在分布式环境中使用,即便是在单机环境中也是不准确的,存在线程安全问题/原子性问题,因此我们一般使用Redis提供的限流模块Redis-Cell来限流,Redis-Cell提供了原子的限流指令cl.throttle,这个留到后续在详细说吧,我要睡觉去了!

以上是关于漏斗限流详述的主要内容,如果未能解决你的问题,请参考以下文章

9.Redis系列Redis的高级应用-漏斗限流

Redis-Cell令牌桶限流详述

微服务网关常用限流算法

限流器系列(3)--自适应限流

限流和常见的三种算法

限流算法介绍