在雙11活動當天凌晨，打折活動開始前多少名客戶下單可以半折甚至是免單優惠，客戶當然不會放過這個一年一次的機會，瘋狂開始。這時候我們程序員小哥哥就苦了，稍一個不注意，服務器駕崩了，次日頭條見。那么為了防止在當天凌晨壓死服務器的并發，我們想到了一個很好的策略，一分鐘搞不定的事情，我們可以兩分鐘搞定，至少保證我們的服務器不會癱瘓，就是說，假如我們的服務器并發量是1w左右，那么我們可以限制在9000的并發量，超過這個預警我們就嚴格限制請求訪問量不能做越過這個警戒線，做到削峰或者說是平滑這個爆發點。

很快我們程序員小哥哥想到了兩個絕佳算法令牌桶和漏斗算法，關于這兩種算法，我們接下來就簡單介紹下。

令牌桶算法：是網絡流量整形和速率限制中最常使用的一種算法。典型情況下，令牌桶算法用來控制發送到網絡上的數據的數目，并允許突發數據的發送。

? ? ? a. 按特定的速率向令牌桶投放令牌

? ? b. 根據預設的匹配規則先對報文進行分類，不符合匹配規則的報文不需要經過令牌桶的處理，直接發送；

? ? c. 符合匹配規則的報文，則需要令牌桶進行處理。當桶中有足夠的令牌則報文可以被繼續發送下去，同時令牌桶中的令牌?量按報文的長度做相應的減少；

? ? d. 當令牌桶中的令牌不足時，報文將不能被發送，只有等到桶中生成了新的令牌，報文才可以發送。這就可以限制報文的流量只能是小于等于令牌生成的速度，達到限制流量的目的。

漏斗算法：主要目的是控制數據注入到網絡的速率，平滑網絡上的突發流量。漏斗算法提供了一種機制，通過它，突發流量可以被整形以便為網絡提供一個穩定的流量

令牌桶和漏斗對比：

兩者主要區別在于漏斗算法能夠強行限制數據的傳輸速率，而令牌桶算法在能夠限制數據的平均傳輸速率外，還允許某種程度的突發傳輸。在令牌桶算法中，只要令牌桶中存在令牌，那么就允許突發地傳輸數據直到達到用戶配置的門限，所以它適合于具有突發特性的流量。

Redis Cell的使用

簡單的介紹這兩種限流算法后，我們今天要來使用的是redis4.0提供的漏斗算法Redis Cell，redis只提供了一個命令cl.throttle

cl.throttle?user?15?30?60?1??????????????▲? ?▲??▲??▲?▲??????????????|???|??|??|?└───?apply?1?token??#每次申請一個token，默認為1??????????????|???|??└──┴─────?30?tokens?/?60?seconds??#速率，60秒生成30個token??????????????|???└───────────?15?max_burst?????#默認最大初始值??????????????└───────────────?key?"user"????#key值 >?cl.throttle?user?15?30?60?11) (integer) 0 # 0 表示允許，1表示拒絕2)?(integer)?15???#?漏斗總容量3)?(integer)?14???#?漏斗剩余空間4) (integer) -1 # 如果拒絕了，需要多長時間后再試(漏斗有空間了，單位秒)5)?(integer)?2?? #?表示多久后令牌桶中的令牌會存滿(單位秒)

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

實現漏斗算法

這里我們簡單的模擬一下漏斗算法

private static class Funnel { private int capacity; private float leakingRate; private int leftQuota; private long leakingTs;? public Funnel(int capacity, int count, int perSecond) { this.capacity = capacity; // 因為計算使用毫秒為單位的 perSecond *= 1000; this.leakingRate = (float) count / perSecond; }? /** * 根據上次水流動的時間，騰出已流出的空間 */ private void makeSpace() { long now = System.currentTimeMillis(); long time = now - leakingTs; int leaked = (int) (time * leakingRate); if (leaked < 1) { return; } leftQuota += leaked; // 如果剩余大于容量，則剩余等于容量 if (leftQuota > capacity) { leftQuota = capacity; } leakingTs = now; }? /**?????????*?漏斗漏水 * @param quota 流量 * @return 是否有足夠的水可以流出（是否允許訪問） */ public boolean watering(int quota) { makeSpace(); int left = leftQuota - quota; if (left >= 0) { leftQuota = left; return true; } return false; } }

漏斗限速方法：

public class FunnelRateLimiter { private Map<String, Funnel> funnelMap = new ConcurrentHashMap<>(); /** * 根據給定的漏斗參數檢查是否允許訪問 * @param username 用戶名 * @param action 操作 * @param capacity 漏斗容量 * @param allowQuota 每單個單位時間允許的流量 * @param perSecond 單位時間（秒） * @return 是否允許訪問 */ public boolean isActionAllowed(String username, String action, int capacity, int allowQuota, int perSecond) { String key = "funnel:" + action + ":" + username; if (!funnelMap.containsKey(key)) { funnelMap.put(key, new Funnel(capacity, allowQuota, perSecond)); } Funnel funnel = funnelMap.get(key); return funnel.watering(1); }}

測試：

public static void main(String[] args) throws InterruptedException { FunnelRateLimiter limiter = new FunnelRateLimiter(); int testAccessCount = 30; int capacity = 5; int allowQuota = 5; int perSecond = 30; int allowCount = 0; int denyCount = 0; for (int i = 0; i < testAccessCount; i++) {????????????boolean?isAllow?=?limiter.isActionAllowed("user",?"doSomething",?5,?5,?30); if (isAllow) { allowCount++; } else { denyCount++; } System.out.println("訪問權限：" + isAllow); Thread.sleep(1000); } System.out.println("報告："); System.out.println("漏斗容量：" + capacity);????????System.out.println("漏斗流動速率："?+?allowQuota?+?"次/"?+?perSecond?+?"秒"); System.out.println("測試次數=" + testAccessCount); System.out.println("允許次數=" + allowCount); System.out.println("拒絕次數=" + denyCount); }

?

一名正在搶救的coder

筆名：mangolove

CSDN地址：https://blog.csdn.net/mango_love

GitHub地址：https://github.com/mangoloveYu

?

總結

以上是生活随笔為你收集整理的Redis高并发限流策略之漏斗限流算法的全部內容，希望文章能夠幫你解決所遇到的問題。

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