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;}//腾出空间太小，最小单位是1if(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分布式锁奥义