百度面試：如何用Redis實現限流？
高并發系統有三大特征：限流、緩存和熔斷，所以限流已經成為當下系統開發中必備的功能了。那么，什么是限流？如何實現限流？使用 Redis 能不能實現限流？接下來我們一起來看。
1.什么是限流？
限流是指在各種應用場景中，通過技術和策略手段對數據流量、請求頻率或資源消耗進行有計劃的限制，以避免系統負載過高、性能下降甚至崩潰的情況發生。限流的目標在于維護系統的穩定性和可用性，并確保服務質量。
使用限流有以下幾個好處：
- 保護系統穩定性：過多的并發請求可能導致服務器內存耗盡、CPU 使用率飽和，從而引發系統響應慢、無法正常服務的問題。
- 防止資源濫用：確保有限的服務資源被合理公平地分配給所有用戶，防止個別用戶或惡意程序過度消耗資源。
- 優化用戶體驗：對于網站和應用程序而言，如果任由高并發導致響應速度變慢，會影響所有用戶的正常使用體驗。
- 保障安全：在網絡層面，限流有助于防范 DoS/DDoS 攻擊，降低系統遭受惡意攻擊的風險。
- 運維成本控制：合理的限流措施可以幫助企業減少不必要的硬件投入，節省運營成本。
2.限流常見算法
限流的常見實現算法有以下幾個：
- 計數器算法：將時間周期劃分為固定大小的窗口（如每分鐘、每小時），并在每個窗口內統計請求的數量。當窗口內的請求數達到預設的閾值時，后續請求將被限制。時間窗口結束后，計數器清零。
- 優點：實現簡單，易于理解。
- 缺點：在窗口切換時刻可能會有突刺流量問題，即在窗口結束時會有短暫的大量請求被允許通過。
- 滑動窗口算法：改進了計算器算法（固定窗口算法）的突刺問題，將時間窗口劃分為多個小的時間段（桶），每個小時間段有自己的計數器。隨著時間流逝，窗口像滑塊一樣平移，過期的小時間段的計數會被丟棄，新時間段加入計數。所有小時間段的計數之和不能超過設定的閾值。
優點：更平滑地處理流量，避免了突刺問題。
缺點：實現相對復雜，需要維護多個計數器。
漏桶算法：想象一個固定容量的桶，水（請求）以恒定速率流入桶中，同時桶底部有小孔讓水以恒定速率流出。當桶滿時，新來的水（請求）會被丟棄。此算法主要用來平滑網絡流量，防止瞬時流量過大。
優點：可以平滑突發流量，保證下游系統的穩定。
缺點：無法處理突發流量高峰，多余的請求會被直接丟棄。
令牌桶算法：與漏桶相反，有一個固定速率填充令牌的桶，令牌代表請求許可。當請求到達時，需要從桶中取出一個令牌，如果桶中有令牌則允許請求通過，否則拒絕。桶的容量是有限的，多余的令牌會被丟棄。
優點：既能平滑流量，又能處理一定程度的突發流量（因為令牌可以累積）。
缺點：需要精確控制令牌生成速度，實現較漏桶復雜。
3.使用Redis實現限流
使用 Redis 也可以實現簡單的限流，它的常見限流方法有以下幾種實現：
- 基于計數器和過期時間實現的計數器算法：使用一個計數器存儲當前請求量（每次使用 incr 方法相加），并設置一個過期時間，計數器在一定時間內自動清零。計數器未到達限流值就可以繼續運行，反之則不能繼續運行。
- 基于有序集合（ZSet）實現的滑動窗口算法：將請求都存入到 ZSet 集合中，在分數（score）中存儲當前請求時間。然后再使用 ZSet 提供的 range 方法輕易的獲取到 2 個時間戳內的所有請求，通過獲取的請求數和限流數進行比較并判斷，從而實現限流。
- 基于列表（List）實現的令牌桶算法：在程序中使用定時任務給 Redis 中的 List 添加令牌，程序通過 List 提供的 leftPop 來獲取令牌，得到令牌繼續執行，否則就是限流不能繼續運行。
了解了以上概念后，接下來我們來看具體的實現。
3.1 計數器算法
此方法的實現思路是：使用一個計數器存儲當前請求量（每次使用 incr 方法相加），并設置一個過期時間，計數器在一定時間內自動清零，從而實現限流。
它的具體操作步驟如下：
- 使用 Redis 的計數器保存當前請求的數量。
- 設置一個過期時間，使得計數器在一定時間內自動清零。
- 每次收到請求時，檢查計數器當前值，如果未達到限流閾值，則增加計數器的值，否則拒絕請求。
具體實現代碼如下：
import redis.clients.jedis.Jedis;
public class RedisRateLimiter {
private static final String REDIS_KEY = "request_counter";
private static final int REQUEST_LIMIT = 100; // 限流閾值
private static final int EXPIRE_TIME = 60; // 過期時間（秒）
public boolean allowRequest() {
Jedis jedis = new Jedis("localhost");
try {
Long counter = jedis.incr(REDIS_KEY);
if (counter == 1) {
// 第一次設置過期時間
jedis.expire(REDIS_KEY, EXPIRE_TIME);
}
if (counter <= REQUEST_LIMIT) {
return true; // 允許請求通過
} else {
return false; // 請求達到限流閾值，拒絕請求
}
} finally {
jedis.close();
}
}
public static void main(String[] args) {
RedisRateLimiter rateLimiter = new RedisRateLimiter();
for (int i = 0; i < 110; i++) {
if (rateLimiter.allowRequest()) {
System.out.println("Request Allowed");
} else {
System.out.println("Request Denied (Rate Limited)");
}
}
}
}在上述代碼中，每次請求會通過 allowRequest() 方法判斷是否達到限流閾值，如果未達到則允許通過，并遞增計數器的值，否則拒絕請求。同時，第一次設置計數器的過期時間，使得計數器在指定的時間內自動清零。
PS：以上是一個簡單的示例，實際應用中需要根據具體場景實現更復雜的限流邏輯，并考慮并發情況下的線程安全性等問題。
因為計算器算法有突刺問題，因此我們需要使用升級版的滑動窗口算法或其他限流算法來解決此問題。
3.2 滑動窗口算法
此方法的實現思路是：將請求都存入到 ZSet 集合中，在分數（score）中存儲當前請求時間。然后再使用 ZSet 提供的 range 方法輕易的獲取到 2 個時間戳內的所有請求，通過獲取的請求數和限流數進行比較并判斷，從而實現限流。
它的具體操作步驟如下：
- 使用有序集合（ZSet）來存儲每個時間窗口內的請求時間戳，成員（member）表示請求的唯一標識，分數（score）表示請求的時間戳。
- 每次收到請求時，將請求的時間戳作為成員，當前時間戳作為分數加入到有序集合中。
- 根據有序集合的時間范圍和滑動窗口的設置，判斷當前時間窗口內的請求數量是否超過限流閾值。
具體實現代碼如下：
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;
import java.util.Set;
public class RedisSlidingWindowRateLimiter {
private static final String ZSET_KEY = "request_timestamps";
private static final int WINDOW_SIZE = 60; // 時間窗口大小（單位：秒）
private static final int REQUEST_LIMIT = 100; // 限流閾值
public boolean allowRequest() {
Jedis jedis = new Jedis("localhost");
long currentTimestamp = System.currentTimeMillis() / 1000;
// 添加當前請求的時間戳到有序集合
jedis.zadd(ZSET_KEY, currentTimestamp, String.valueOf(currentTimestamp));
// 移除過期的請求時間戳，保持時間窗口內的請求
long start = currentTimestamp - WINDOW_SIZE;
long end = currentTimestamp;
jedis.zremrangeByScore(ZSET_KEY, 0, start);
// 查詢當前時間窗口內的請求數量
Set<Tuple> requestTimestamps = jedis.zrangeByScoreWithScores(ZSET_KEY, start, end);
long requestCount = requestTimestamps.size();
jedis.close();
// 判斷請求數量是否超過限流閾值
return requestCount <= REQUEST_LIMIT;
}
public static void main(String[] args) {
RedisSlidingWindowRateLimiter rateLimiter = new RedisSlidingWindowRateLimiter();
for (int i = 0; i < 110; i++) {
if (rateLimiter.allowRequest()) {
System.out.println("Request Allowed");
} else {
System.out.println("Request Denied (Rate Limited)");
}
}
}
}在上述代碼中，每次收到請求時，將當前請求的時間戳加入到有序集合中，并移除過期的請求時間戳，然后查詢當前時間窗口內的請求數量，判斷是否達到限流閾值。這樣基于 Redis 的滑動窗口限流算法可以有效控制單位時間內的請求流量，避免系統被過多請求壓垮。
3.3 令牌桶算法
此方法的實現思路是：在程序中使用定時任務給 Redis 中的 List 添加令牌，程序通過 List 提供的 leftPop 來獲取令牌，得到令牌繼續執行，否則就是限流不能繼續運行。
① 添加令牌
在 Spring Boot 項目中，通過定時任務給 Redis 中的 List 每秒中添加一個令牌（當然也可以通過修改定時任務的執行時間來控制令牌的發放速度），具體實現代碼如下：
@Configuration // 1.注入到 IoC 中，啟動程序時加載
@EnableScheduling // 2.開啟定時任務
public class SaticScheduleTask {
@Autowired
private RedisTemplate redisTemplate;
// 3.添加定時任務
@Scheduled(fixedRate = 1000)
private void configureTasks() {
redisTemplate.opsForList().rightPush("limit_list",UUID.randomUUID().toString());
}
}② 獲取令牌
令牌的獲取代碼如下：
public boolean allowRequest(){
Object result = redisTemplate.opsForList().leftPop("limit_list");
if(result == null){
return false;
}
return true;
}在上述代碼中，我們每次訪問 allowRequest() 方法時，會嘗試從 Redis 中獲取一個令牌，如果拿到令牌了，那就說明沒超出限制，可以繼續執行，反之則不能執行。
