Java 對IP請求進行限流.

高並發系統下, 有三把利器 緩存 降級 限流.

• 緩存: 將常用數據緩存起來, 減少數據庫或者磁盤IO
• 降級: 保護核心系統, 降低非核心業務請求響應
• 限流: 在某一個時間窗口內對請求進行限速, 保護系統

 本文主要介紹限流, 常見限流算法中又分為計數器算法, 漏桶算法, 令牌桶算法.

計數器算法

比較簡單, 直接用一個map + counter即可實現. 請求來了, 以IP為key,

查詢下之前響應次數, 如果調用次數超出MAX_COUT, 返回失敗, 屬於簡單粗暴型選手.

漏桶算法

請求全部進入漏桶, 漏桶恆定速率輸出反饋. 這樣可以保證數據傳輸平滑,

但是無法預防突發大量請求, 一秒來了100個請求, 都要阻塞排隊, 從小水管輸出數據.

 

令牌桶算法

令牌桶是以固定速度往桶里存令牌, 例如一秒存1000個令牌, 業務請求來了, 直接從桶里獲取令牌響應輸出.

跟漏桶的差異在於, 他可以預存令牌, 如果一秒鍾來了100個請求, 桶里有100個令牌,

那么可以立刻響應給客戶端, 而不是排隊輸出.

 

令牌桶的實現

guava中提供了令牌桶的一個封裝實現RateLimiter, 可以直接調用, 省的我們自己包裝ConcurrentHashMap + Timer.

我們預設的場景是服務器端提供一個API供不同客戶端查詢, 要限流每個IP每秒只能調用兩次該API.

首先要定義一個服務器端的緩存, 定期清理即可, 緩存 IP : 令牌桶

    // 根據IP分不同的令牌桶, 每天自動清理緩存
    private static LoadingCache<String, RateLimiter> caches = CacheBuilder.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(1, TimeUnit.DAYS)
            .build(new CacheLoader<String, RateLimiter>() {
                @Override
                public RateLimiter load(String key) throws Exception {
                    // 新的IP初始化 (限流每秒兩個令牌響應)
                    return RateLimiter.create(2);
                }
            });

然后在業務代碼中進行限流調用

    private static void login(int i) throws ExecutionException {
        // 模擬IP的key
        String ip = String.valueOf(i).charAt(0) + "";
        RateLimiter limiter = caches.get(ip);

        if (limiter.tryAcquire()) {
            System.out.println(i + " success " + new SimpleDateFormat("HH:mm:ss.sss").format(new Date()));
        } else {
            System.out.println(i + " failed " + new SimpleDateFormat("HH:mm:ss.sss").format(new Date()));
        }
    }

模擬客戶端調用

        for (int i = 0; i < 1000; i++) {
            // 模擬實際業務請求
            Thread.sleep(100);
            login(i);
        }

完整代碼

public class doLimit {

    // 根據IP分不同的令牌桶, 每天自動清理緩存
    private static LoadingCache<String, RateLimiter> caches = CacheBuilder.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(1, TimeUnit.DAYS)
            .build(new CacheLoader<String, RateLimiter>() {
                @Override
                public RateLimiter load(String key) throws Exception {
                    // 新的IP初始化 (限流每秒兩個令牌響應)
                    return RateLimiter.create(2);
                }
            });

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        for (int i = 0; i < 1000; i++) {
            // 模擬實際業務請求
            Thread.sleep(100);
            login(i);
        }
    }

    private static void login(int i) throws ExecutionException {
        // 模擬IP的key
        String ip = String.valueOf(i).charAt(0) + "";
        RateLimiter limiter = caches.get(ip);

        if (limiter.tryAcquire()) {
            System.out.println(i + " success " + new SimpleDateFormat("HH:mm:ss.sss").format(new Date()));
        } else {
            System.out.println(i + " failed " + new SimpleDateFormat("HH:mm:ss.sss").format(new Date()));
        }
    }
}