Redis流量控制策略

怼烎@ 2023-02-14 15:55 21阅读 0赞

#### Reids 简单流控 ####

*  流控是分布式领域一个被经常用到的一个计数，当系统承载能力有限的时候，如何组织计划外的请求继续对系统施加压力，这是一个需要解决的问题，在系统承载达到峰值的时候，我们需要弃车保帅，保证主流程业务的通畅，除了流控，限流还有一个目的，控制用户行为，避免垃圾请求以及屏蔽某些爬虫软件爬取数据。

##### 如何使用Redis进行流控 #####

*  一个简单的，常见的案例。系统需要限定某个用户的某个行为在指定时间内只能允许发生N次，如何使用Redis的数据结构来实现，我们限定义如下一个接口：

//指定用户userId， 某个接口（行为）actionKey，指定时间time， 请求次数maxcount
    boolean iaActionAllowed(Long userId, String actionKey, Long time, Long maxCount);

*  通过以上我们可以得出一个简单的实现方案，在接口A内，每次userId请求，将key进行累加，并且设置key过期时间time，当value > maxCount 则阻断。如下简单版本的流控：

public class SimpleRateLimit { 
        public static Long time = 30L;
        public static Boolean isActionAllowed(Long userId, String actionKey, Long maxCount) { 
            String key = actionKey + "_" + userId.toString();
            Jedis jedis = JedisUtils.getJedis();
            if (jedis.exists(key)) { 
                Long times = jedis.incr(key);
                return times > maxCount;
            }else { 
                jedis.set(key, "1");
                jedis.expire(key, time.intValue());
            }
            return true;
        }
        public static void main(String[] args) { 
            for (int i = 0; i < 100; i++) { 
                if(isActionAllowed(123L, "login", 30L)){ 
                    System.out.println("allowed this action login : " + i);
                }else { 
                    System.out.println("close the door : "+ i);
                }
            }
        }
    }

*  以上是一个简单版本的流控，只能正对某一个接口功能进行控制，但是这种方式也有一定的弊端，就是我们每次请求都需要判断，判断，虽然Redis的性能比较高，但是每个接口都需要过一次这个逻辑，并且不同的接口我们要用不同的key，整体流控也需要做另外的key，就是还有一定的提升空间了

##### 漏斗限流 #####

*  漏斗限流是最常用的流控方法，他相关的算法有令牌桶算法，漏桶算法。
 *  如下面图中所示，漏斗的容量是有先的，如果将漏嘴堵住，一直灌水，就满了，直到装不进去。打开漏嘴，水下流。灌水速度大于流速，满了就需要等待，灌水速度小于流速，永远也满不了。所以漏斗的剩余空间代表这当前行为可以持续进行的数量，漏嘴的流速代表系统允许该行为的最大频率。
 *  或者这样理解，向漏斗加水相当于添加令牌，每次请求需要获取一个令牌，漏斗中令牌数相当于系统当前行为可以持续进行的数量，获取令牌的速度代表系统该欣慰的最大频率。如下代码

/** * 有点像令牌桶的漏桶 * @author liaojiamin * @Date:Created in 16:57 2020/5/29 */
    public class FunnelRateLimiter { 
        private Map<String, Funnel> funnels = new HashMap<>();
    
        public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate){ 
            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.watering(1);
        }
        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);
                //int类型越界情况 重新初始化
                if(deltaQuota < 0){ 
                    this.leftQuota = capacity;
                    this.leakingTs = nowTs;
                    return;
                }
                //腾出空间太小，最小单位是1
                if(deltaQuota < 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;
            }
        }
    }

*  如上代码，funnel对象的Mask\_space是令牌桶核心算法，每次消耗之前都会计算一次桶中的令牌，一次可能消耗一个令牌，当并发请求，一秒可能消耗多个令牌，计算桶中令牌数根据时间固定速率添加，最大值capacity。这样就有一个令牌桶算法。
 *  那么我们怎么用Redis来实现一个令牌桶，有Redis的数据结构能搞定没。
 *  我们可以用hash结构，添加令牌时候将Hash字段取出运算，在更新，这个问题在于非原子操作，非线程安全的。如果考虑事务，就有失败，重试的情况使代码变得更加复杂，得不偿失。

##### Redis-Cell #####

*  Redis 4.0 提供来一个流控的Redis模块，他叫Redis-Cell。这个模块也用了漏斗算法，并提供来原子的限流指令，有了这个模块，限流可以通过简单命令来解决。
 *  该模块只有一个指令cl.throttle，他的参数和返回值都比较复杂，如下使用方式：

cl.throttle jiamin:reply 15 30 60 1

*  如上，加入令牌频率为60s最多30个，漏斗初始容量15个令牌，也就是说开始可以连续15个请求，然后才开始受到添加令牌的速率影响。此处令牌添加的速度变成两个参数，替代之前的单个浮点数。用两个参数相除的结果来表示添加令牌的速度更直观。
 *  返回值有几种：
    
     *  0: 表示允许
     *  1:表示拒绝
     *  15 ：漏桶中令牌容量capacity
     *  14: 漏桶中剩余令牌 left\_quota
     *  \-1 如果被拒绝，需要多长时间后再试
     *  2：多长时间后，漏斗完全填满令牌：left\_quota == capacity 单位秒
 *  在执行限流指令时候，如果被拒绝，就需要丢弃或者重试，cl.throttle指令考虑得非常周到，连重试的时间都帮你算好了，直接取返回结果数组的第四个值进行sleep既可，如果不想苏泽线程，也可以异步定时任务来重试。

[上一篇Redis高级数据结构][Redis]  
[下一篇Redis分布式锁奥义][Redis 1]

[Redis]: https://blog.csdn.net/liaojiamin0102/article/details/106492905
[Redis 1]: https://blog.csdn.net/liaojiamin0102/article/details/106539873