Reids 簡單流控

• 流控是分布式領域一個被經(jīng)常用到的一個計數(shù)，當系統(tǒng)承載能力有限的時候，如何組織計劃外的請求繼續(xù)對系統(tǒng)施加壓力，這是一個需要解決的問題，在系統(tǒng)承載達到峰值的時候，我們需要棄車保帥，保證主流程業(yè)務的通暢，除了流控，限流還有一個目的，控制用戶行為，避免垃圾請求以及屏蔽某些爬蟲軟件爬取數(shù)據(jù)。

如何使用Redis進行流控

• 一個簡單的，常見的案例。系統(tǒng)需要限定某個用戶的某個行為在指定時間內只能允許發(fā)生N次，如何使用Redis的數(shù)據(jù)結構來實現(xiàn)，我們限定義如下一個接口：

//指定用戶userId， 某個接口（行為）actionKey，指定時間time， 請求次數(shù)maxcount boolean iaActionAllowed(Long userId, String actionKey, Long time, Long maxCount);

• 通過以上我們可以得出一個簡單的實現(xiàn)方案，在接口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，就是還有一定的提升空間了

漏斗限流

• 漏斗限流是最常用的流控方法，他相關的算法有令牌桶算法，漏桶算法。
• 如下面圖中所示，漏斗的容量是有先的，如果將漏嘴堵住，一直灌水，就滿了，直到裝不進去。打開漏嘴，水下流。灌水速度大于流速，滿了就需要等待，灌水速度小于流速，永遠也滿不了。所以漏斗的剩余空間代表這當前行為可以持續(xù)進行的數(shù)量，漏嘴的流速代表系統(tǒng)允許該行為的最大頻率。
• 或者這樣理解，向漏斗加水相當于添加令牌，每次請求需要獲取一個令牌，漏斗中令牌數(shù)相當于系統(tǒng)當前行為可以持續(xù)進行的數(shù)量，獲取令牌的速度代表系統(tǒng)該欣慰的最大頻率。如下代碼

/*** 有點像令牌桶的漏桶* @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;//漏桶現(xiàn)有配額int leftQuota;//開始時間long leakingTs;public Funnel(int capacity, float leakingRate){this.capacity = capacity;this.leakingRate = leakingRate;this.leftQuota = capacity;this.leakingTs = System.currentTimeMillis();}//獲取空間（更新當前桶中的令牌數(shù)量，依據(jù)時間以及流速）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是令牌桶核心算法，每次消耗之前都會計算一次桶中的令牌，一次可能消耗一個令牌，當并發(fā)請求，一秒可能消耗多個令牌，計算桶中令牌數(shù)根據(jù)時間固定速率添加，最大值capacity。這樣就有一個令牌桶算法。
• 那么我們怎么用Redis來實現(xiàn)一個令牌桶，有Redis的數(shù)據(jù)結構能搞定沒。
• 我們可以用hash結構，添加令牌時候將Hash字段取出運算，在更新，這個問題在于非原子操作，非線程安全的。如果考慮事務，就有失敗，重試的情況使代碼變得更加復雜，得不償失。

Redis-Cell

• Redis 4.0 提供來一個流控的Redis模塊，他叫Redis-Cell。這個模塊也用了漏斗算法，并提供來原子的限流指令，有了這個模塊，限流可以通過簡單命令來解決。
• 該模塊只有一個指令cl.throttle，他的參數(shù)和返回值都比較復雜，如下使用方式：

cl.throttle jiamin:reply 15 30 60 1

• 如上，加入令牌頻率為60s最多30個，漏斗初始容量15個令牌，也就是說開始可以連續(xù)15個請求，然后才開始受到添加令牌的速率影響。此處令牌添加的速度變成兩個參數(shù)，替代之前的單個浮點數(shù)。用兩個參數(shù)相除的結果來表示添加令牌的速度更直觀。

• 返回值有幾種：

  • 0: 表示允許
  • 1:表示拒絕
  • 15 ：漏桶中令牌容量capacity
  • 14: 漏桶中剩余令牌 left_quota
  • -1 如果被拒絕，需要多長時間后再試
  • 2：多長時間后，漏斗完全填滿令牌：left_quota == capacity 單位秒

• 在執(zhí)行限流指令時候，如果被拒絕，就需要丟棄或者重試，cl.throttle指令考慮得非常周到，連重試的時間都幫你算好了，直接取返回結果數(shù)組的第四個值進行sleep既可，如果不想蘇澤線程，也可以異步定時任務來重試。

上一篇Redis高級數(shù)據(jù)結構
下一篇Redis分布式鎖奧義

總結

以上是生活随笔為你收集整理的Redis流量控制策略的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內容還不錯，歡迎將生活随笔推薦給好友。