精講Redis限流：多種方法與生產實踐
令牌桶算法實現限流
令牌桶算法是一種常見的限流算法，它通過維護一個固定容量的令牌桶來控制流量。每個請求需要獲取一個令牌，如果桶中沒有足夠的令牌，則請求會被限制。
首先，你需要在Redis中設置一個計數器和一個定時器來模擬令牌桶：
import redis
import time
# 連接Redis
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 設置令牌桶容量和每秒生成的令牌數
bucket_capacity = 10
tokens_per_second = 2
# 初始化令牌桶
r.set('tokens', bucket_capacity)
r.set('last_time', int(time.time()))
# 請求令牌的函數
def request_token():
current_time = int(time.time())
last_time = int(r.get('last_time'))
elapsed_time = current_time - last_time
# 計算新增的令牌數量
new_tokens = elapsed_time * tokens_per_second
current_tokens = int(r.get('tokens'))
# 更新令牌數量
if new_tokens + current_tokens > bucket_capacity:
r.set('tokens', bucket_capacity)
else:
r.set('tokens', new_tokens + current_tokens)
r.set('last_time', current_time)
# 使用令牌的代碼
def process_request():
if int(r.get('tokens')) > 0:
# 執行你的請求處理邏輯
print('請求通過')
r.decr('tokens') # 消耗一個令牌
else:
print('請求被限制')
# 測試請求
for _ in range(15):
request_token()
process_request()
time.sleep(1)這個示例中，我們通過Redis來維護令牌桶的狀態，并在請求到來時檢查是否有足夠的令牌。如果有足夠的令牌，請求將被處理，否則請求將被限制。
漏桶算法實現限流
漏桶算法是另一種流量控制算法，它維護一個固定容量的漏桶，請求進來后，會以固定速率從漏桶中排出。
以下是使用Redis實現漏桶算法的示例：
import redis
import time
# 連接Redis
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 設置漏桶容量和漏出速率（每秒排出的請求數）
bucket_capacity = 10
leak_rate = 2
# 初始化漏桶
r.set('bucket_capacity', bucket_capacity)
r.set('last_leak_time', int(time.time()))
# 請求處理函數
def process_request():
current_time = int(time.time())
last_leak_time = int(r.get('last_leak_time'))
time_elapsed = current_time - last_leak_time
# 計算漏出的請求數
leaked_requests = min(int(r.get('bucket_capacity')) * (time_elapsed // 1), int(r.get('bucket_capacity')))
# 更新漏桶狀態
r.incrby('bucket_capacity', leaked_requests)
r.set('last_leak_time', current_time)
# 處理請求
if int(r.get('bucket_capacity')) >= 1:
print('請求通過')
r.decr('bucket_capacity')
else:
print('請求被限制')
# 測試請求
for _ in range(15):
process_request()
time.sleep(1)在漏桶算法中，請求會被排入漏桶中，然后以固定速率漏出。如果漏桶中有請求，則請求會被處理，否則請求會被限制。
以上兩個案例雖然能夠實現限流，但是存在一定的問題，無法滿足生產的要求，下面講一下其他思路。
有序集合zset實現限流
使用Redis的有序集合（ZSET）也可以實現限流功能。有序集合中的成員可以關聯一個分數，我們可以使用分數來表示每個請求的權重或時間戳，并利用有序集合的排序特性來判斷請求是否被允許。
以下是使用有序集合實現基于時間窗口的限流示例：
import redis
import time
# 連接Redis
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 限流配置
max_requests = 10 # 在時間窗口內允許的最大請求數
window_duration = 60 # 時間窗口的持續時間（秒）
# 請求處理函數
def process_request(user_id):
current_time = time.time()
zset_key = "requests:" + user_id
# 刪除時間窗口之外的請求記錄
r.zremrangebyscore(zset_key, '-inf', current_time - window_duration)
# 獲取當前時間窗口內的請求數
requests_in_window = r.zcard(zset_key)
if requests_in_window < max_requests:
# 如果請求數在限制范圍內，允許請求并記錄請求時間
r.zadd(zset_key, {str(current_time): current_time})
print('請求通過')
else:
print('請求被限制')
# 測試請求
user_id = "user123"
for _ in range(15):
process_request(user_id)
time.sleep(2)在這個示例中，我們為每個用戶維護一個有序集合，其中成員是請求的時間戳，分數也設置為時間戳。在處理請求時，我們首先刪除時間窗口之外的請求記錄，然后檢查時間窗口內的請求數是否超過了限制。如果沒有超過限制，允許請求并記錄請求時間戳。
這種方法可以實現基于時間窗口的限流，你可以根據需要調整max_requests和window_duration來配置限流策略。
但是這樣又引發了一個并發性問題
在分布式系統中，處理請求的并發性是一個重要考慮因素，特別是在多個客戶端同時發送請求的情況下。以下是一些常見的方法來確保process_request操作的并發安全性：
互斥鎖（Mutex Lock）：使用互斥鎖可以確保在同一時刻只有一個線程或進程可以執行process_request操作。這可以通過在關鍵部分的代碼周圍放置鎖來實現。在Redis中，你可以使用Redis的SETNX（Set If Not Exists）命令來實現互斥鎖，確保只有一個客戶端可以獲取鎖并執行請求處理操作。
def process_request(user_id):
lock_key = "lock:" + user_id
acquired_lock = r.setnx(lock_key, "1")
if acquired_lock:
try:
# 在獲取鎖后，執行請求處理操作
current_time = time.time()
zset_key = "requests:" + user_id
# 刪除時間窗口之外的請求記錄
r.zremrangebyscore(zset_key, '-inf', current_time - window_duration)
# 獲取當前時間窗口內的請求數
requests_in_window = r.zcard(zset_key)
if requests_in_window < max_requests:
# 如果請求數在限制范圍內，允許請求并記錄請求時間
r.zadd(zset_key, {str(current_time): current_time})
print('請求通過')
else:
print('請求被限制')
finally:
# 釋放鎖
r.delete(lock_key)
else:
print('無法獲取鎖，請求被限制')分布式鎖：如果你的系統是分布式的，你可以考慮使用分布式鎖來確保不同節點上的請求處理代碼不會同時執行。一些常見的分布式鎖實現包括基于ZooKeeper或Redis的分布式鎖。這些鎖可以協調不同節點之間的并發執行。
事務：Redis支持事務，你可以使用MULTI和EXEC命令將多個操作包裝在一個事務中。在這種情況下，Redis會確保整個事務要么全部成功執行，要么全部失敗，從而保證一致性。
這種雖然能解決問題，但是并不是最優解
EXEC + lua 實現
使用Redis的EXEC命令和Lua腳本可以確保多個Redis命令在一個事務中執行，從而保證一致性。下面是一個使用EXEC和Lua腳本來實現請求處理的示例：
import redis
# 連接Redis
r = redis.StrictRedis(host='localhost', port=6379, db=0)
# 限流配置
max_requests = 10 # 在時間窗口內允許的最大請求數
window_duration = 60 # 時間窗口的持續時間（秒）
# Lua腳本，用于限流處理
lua_script = """
local user_id = KEYS[1]
local max_requests = tonumber(ARGV[1])
local window_duration = tonumber(ARGV[2])
local current_time = tonumber(ARGV[3])
-- 刪除時間窗口之外的請求記錄
redis.call('ZREMRANGEBYSCORE', 'requests:'..user_id, '-inf', current_time - window_duration)
-- 獲取當前時間窗口內的請求數
local requests_in_window = redis.call('ZCARD', 'requests:'..user_id)
if requests_in_window < max_requests then
-- 如果請求數在限制范圍內，允許請求并記錄請求時間
redis.call('ZADD', 'requests:'..user_id, current_time, current_time)
return 'ALLOWED'
else
return 'LIMITED'
end
"""
# 請求處理函數
def process_request(user_id):
current_time = int(time.time())
result = r.eval(lua_script, 1, user_id, max_requests, window_duration, current_time)
if result == b'ALLOWED':
print('請求通過')
else:
print('請求被限制')
# 測試請求
user_id = "user123"
for _ in range(15):
process_request(user_id)
time.sleep(2)在上述示例中，我們使用Lua腳本編寫了一個與之前的請求處理邏輯相同的限流處理邏輯。然后，我們通過eval命令將Lua腳本傳遞給Redis，并在一個事務中執行它。這樣可以確保在同一事務內執行多個Redis命令，從而保證了一致性。
請注意，在Lua腳本中，我們使用了Redis的命令來執行限流邏輯，然后根據結果返回相應的值，以便在Python中進行處理。如果請求被限制，Lua腳本返回'LIMITED'，否則返回'ALLOWED'。
通過這種方式，你可以使用Redis實現基于令牌桶算法的限流功能。可以根據需要調整令牌桶容量和生成速率來滿足你的應用需求。此外，需要注意在高并發情況下，需要謹慎處理并發問題。
