共計 3762 個字符，預計需要花費 10 分鐘才能閱讀完成。
丸趣 TV 小編給大家分享一下利用 Redis 如何實現令牌桶算法，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！
在限流算法中有一種令牌桶算法，該算法可以應對短暫的突發流量，這對于現實環境中流量不怎么均勻的情況特別有用，不會頻繁的觸發限流，對調用方比較友好。
例如，當前限制 10qps，大多數情況下不會超過此數量，但偶爾會達到 30qps，然后很快就會恢復正常，假設這種突發流量不會對系統穩定性產生影響，我們可以在一定程度上允許這種瞬時突發流量，從而為用戶帶來更好的可用性體驗。這就是使用令牌桶算法的地方。
令牌桶算法原理
如下圖所示，該算法的基本原理是：有一個容量為 X 的令牌桶，每 Y 單位時間內將 Z 個令牌放入該桶。如果桶中的令牌數量超過 X，那么它將被丟棄。處理請求時，需要先從令牌桶中取出令牌，如果拿到了令牌，則繼續處理；如果拿不到令牌，則拒絕請求。
可以看出，在令牌桶算法中設置 X，Y 和 Z 的數量尤為重要。Z 應該比每 Y 單位時間內的請求數稍大，系統將長時間處于此狀態；X 是系統允許的瞬時最大請求數，并且系統不應該長時間處于此狀態，否則就會頻繁觸發限流，此時表明流量出現了超預期的情況，需要及時調查原因并采取相應措施。
Redis 實現令牌桶算法
之前看過有些程序實現的令牌桶，其向桶中放入令牌的方法是啟動一個線程，每隔 Y 單位時間增加一次令牌數量，或者在 Timer 中定時執行這一過程。我不太滿意這種方法，原因有二，一是浪費線程資源，二是因為調度的問題執行時間不精確。【相關推薦：Redis 視頻教程】
這里確定令牌桶中令牌數量的方法是通過計算得出，首先算出從上次請求到這次請求經過了多長時間，是否達到發令牌的時間閾值，然后增加的令牌數是多少，這些令牌能夠放到桶中的是多少。
Talk is cheap!
下邊就來看看 Redis 中怎么實現的，因為涉及到多次與 Redis 的交互，這里為了提高限流處理的吞吐量，減少程序與 Redis 的交互次數，采用了 Redis 支持的 Lua script，Lua script 的執行是原子的，所以也不用擔心出現臟數據的問題。
代碼節選自 FireflySoft.RateLimit，它不僅支持普通主從部署 Redis，還支持集群 Redis，所以吞吐量可以通過水平擴展的方式進行提升。為了方便閱讀，這里增加一些注釋，實際是沒有的。
-- 定義返回值，是個數組，包含：是否觸發限流（1 限流 0 通過）、當前桶中的令牌數
local ret={}
ret[1]=0
-- Redis 集群分片 Key，KEYS[1]是限流目標
local cl_key = { .. KEYS[1] .. }
-- 獲取限流懲罰的當前設置，觸發限流懲罰時會寫一個有過期時間的 KV
-- 如果存在限流懲罰，則返回結果[1,-1]
local lock_key=cl_key .. -lock
local lock_val=redis.call(get ,lock_key)
if lock_val == 1 then
ret[1]=1
ret[2]=-1
return ret;
-- 這里省略部分代碼
-- 獲取[上次向桶中投放令牌的時間]，如果沒有設置過這個投放時間，則令牌桶也不存在，此時：-- 一種情況是：首次執行，此時定義令牌桶就是滿的。-- 另一種情況是：較長時間沒有執行過限流處理，導致承載這個時間的 KV 被釋放了，-- 這個過期時間會超過自然投放令牌到桶中直到桶滿的時間，所以令牌桶也應該是滿的。local last_time=redis.call(get ,st_key)
if(last_time==false)
-- 本次執行后剩余令牌數量：桶的容量 - 本次執行消耗的令牌數量
bucket_amount = capacity - amount;
-- 將這個令牌數量更新到令牌桶中，同時這里有個過期時間，如果長時間不執行這個程序，令牌桶 KV 會被回收
redis.call(set ,KEYS[1],bucket_amount, PX ,key_expire_time)
-- 設置[上次向桶中放入令牌的時間]，后邊計算應放入桶中的令牌數量時會用到
redis.call(set ,st_key,start_time, PX ,key_expire_time)
-- 返回值[當前桶中的令牌數]
ret[2]=bucket_amount
-- 無需其它處理
return ret
-- 令牌桶存在，獲取令牌桶中的當前令牌數
local current_value = redis.call(get ,KEYS[1])
current_value = tonumber(current_value)
-- 判斷是不是該放入新令牌到桶中了：當前時間 - 上次投放的時間 = 投放的時間間隔
last_time=tonumber(last_time)
local last_time_changed=0
local past_time=current_time-last_time
if(past_time inflow_unit)
-- 不到投放的時候，直接從令牌桶中取走令牌
bucket_amount=current_value-amount
-- 需要放入一些令牌, 預計投放數量 = (距上次投放過去的時間 / 投放的時間間隔)* 每單位時間投放的數量
local past_inflow_unit_quantity = past_time/inflow_unit
past_inflow_unit_quantity=math.floor(past_inflow_unit_quantity)
last_time=last_time+past_inflow_unit_quantity*inflow_unit
last_time_changed=1
local past_inflow_quantity=past_inflow_unit_quantity*inflow_quantity_per_unit
bucket_amount=current_value+past_inflow_quantity-amount
-- 這里省略部分代碼
ret[2]=bucket_amount
-- 如果桶中剩余數量小于 0，則看看是否需要限流懲罰，如果需要則寫入一個懲罰 KV，過期時間為懲罰的秒數
if(bucket_amount 0)
if lock_seconds 0 then
redis.call(set ,lock_key, 1 , EX ,lock_seconds, NX)
end
ret[1]=1
return ret
-- 來到這里，代表可以成功扣減令牌，則需要更新令牌桶 KV
if last_time_changed==1 then
redis.call(set ,KEYS[1],bucket_amount, PX ,key_expire_time)
-- 有新投放，更新 [上次投放時間] 為本次投放時間
redis.call(set ,st_key,last_time, PX ,key_expire_time)
redis.call(set ,KEYS[1],bucket_amount, PX ,key_expire_time)
return ret通過以上代碼，可以看出，其主要處理過程是：
1、判斷有沒有被限流懲罰，有則直接返回，無則進入下一步。
2、判斷令牌桶是否存在，不存在則先創建令牌桶，然后扣減令牌返回，存在則進入下一步。
3、判斷是否需要投放令牌，不需要則直接扣減令牌，需要則先投放令牌再扣減令牌。
4、判斷扣減后的令牌數，如果小于 0 則返回限流，同時設置限流懲罰，如果大于等于 0 則進入下一步。
5、更新桶中的令牌數到 Redis。
你可以在任何一種開發語言的 Redis 庫中提交并運行這段 Lua script 腳本，如果你使用的是.NET 平臺，可以參考這篇文章：ASP.NET Core 中使用令牌桶限流（https://blog.bossma.cn/dotnet/asp-net-core-token-bucket-algorithm-of-rate-limit/）。
關于 FireflySoft.RateLimit
FireflySoft.RateLimit 是一個基于 .NET Standard 的限流類庫，其內核簡單輕巧，能夠靈活應對各種需求的限流場景。
其主要特點包括：
多種限流算法：內置固定窗口、滑動窗口、漏桶、令牌桶四種算法，還可自定義擴展。
多種計數存儲：目前支持內存、Redis 兩種存儲方式。
分布式友好：通過 Redis 存儲支持分布式程序統一計數。
限流目標靈活：可以從請求中提取各種數據用于設置限流目標。
支持限流懲罰：可以在客戶端觸發限流后鎖定一段時間不允許其訪問。
動態更改規則：支持程序運行時動態更改限流規則。
自定義錯誤：可以自定義觸發限流后的錯誤碼和錯誤消息。
普適性：原則上可以滿足任何需要限流的場景。
以上是“利用 Redis 如何實現令牌桶算法”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注丸趣 TV 行業資訊頻道！
