Java高並行系統限流演演算法的實現

2022-05-18 10:00:19

1 概述

在開發高並行系統時有三把利器用來保護系統：快取、降級和限流。限流可以認為服務降級的一種，限流是對系統的一種保護措施。即限制流量請求的頻率（每秒處理多少個請求）。一般來說，當請求流量超過系統的瓶頸，則丟棄掉多餘的請求流量，保證系統的可用性。即要麼不放進來，放進來的就保證提供服務。比如：延遲處理，拒絕處理，或者部分拒絕處理等等。

2 計數器限流

2.1 概述

計數器採用簡單的計數操作，到一段時間節點後自動清零

2.2 實現

package com.oldlu.limit;
import java.util.concurrent.*;
public class Counter {
    public static void main(String[] args) {
        //計數器，這裡用號誌實現
        final Semaphore semaphore = new Semaphore(3);
        //定時器，到點清零
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                semaphore.release(3);
            }
        },3000,3000,TimeUnit.MILLISECONDS);
        //模擬無數個請求從天而降
        while (true) {
            try {
                //判斷計數器
                semaphore.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //如果准許響應，列印一個ok
            System.out.println("ok");
        }
    }
}

2.3 結果分析

3個ok一組呈現，到下一個計數週期之前被阻斷

2.4 優缺點

實現起來非常簡單。
控制力度太過於簡略，假如1s內限制3次，那麼如果3次在前100ms內已經用完，後面的900ms將只能處於阻塞狀態，白白浪費掉。

2.5 應用

使用計數器限流的場景較少，因為它的處理邏輯不夠靈活。最常見的可能在web的登入密碼驗證，輸入錯誤次數凍結一段時間的場景。如果網站請求使用計數器，那麼惡意攻擊者前100ms吃掉流量計數，使得後續正常的請求被全部阻斷，整個服務很容易被搞垮。

3 漏桶演演算法

3.1 概述

漏桶演演算法將請求快取在桶中，服務流程勻速處理。超出桶容量的部分丟棄。漏桶演演算法主要用於保護內部的處理業務，保障其穩定有節奏的處理請求，但是無法根據流量的波動彈性調整響應能力。現實中，類似容納人數有限的服務大廳開啟了固定的服務視窗。

3.2 實現

package com.oldlu.limit;
import java.util.concurrent.*;
public class Barrel {
    public static void main(String[] args) {
        //桶，用阻塞佇列實現，容量為3
        final LinkedBlockingQueue<Integer> que = new LinkedBlockingQueue(3);
        //定時器，相當於服務的視窗，2s處理一個
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                int v = que.poll();
                System.out.println("處理："+v);
            }
        },2000,2000,TimeUnit.MILLISECONDS);
        //無數個請求，i 可以理解為請求的編號
        int i=0;
        while (true) {
            i++;
            try {
                System.out.println("put:"+i);
                //如果是put，會一直等待桶中有空閒位置，不會丟棄
//                que.put(i);
                //等待1s如果進不了桶，就溢位丟棄
                que.offer(i,1000,TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

3.3 結果分析

put任務號按照順序入桶
執行任務勻速的1s一個被處理
因為桶的容量只有3，所以1-3完美執行，4被溢位丟棄，5正常執行

3.4 優缺點

有效的擋住了外部的請求，保護了內部的服務不會過載
內部服務勻速執行，無法應對流量洪峰，無法做到彈性處理突發任務
任務超時溢位時被丟棄。現實中可能需要快取佇列輔助保持一段時間
5）應用
nginx中的限流是漏桶演演算法的典型應用，設定案例如下：

http {
    #$binary_remote_addr 表示通過remote_addr這個標識來做key，也就是限制同一使用者端ip地址。
#zone=one:10m 表示生成一個大小為10M，名字為one的記憶體區域，用來儲存存取的頻次資訊。    
#rate=1r/s 表示允許相同標識的使用者端每秒1次存取    
    limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;
    server {
        location /limited/ {
        #zone=one 與上面limit_req_zone 裡的name對應。
#burst=5 緩衝區，超過了存取頻次限制的請求可以先放到這個緩衝區內，類似程式碼中的佇列長度。        
#nodelay 如果設定，超過存取頻次而且緩衝區也滿了的時候就會直接返回503，如果沒有設定，則所有請求
會等待排隊，類似程式碼中的put還是offer。  
       
        limit_req zone=one burst=5 nodelay;
    }
}

4 令牌桶演演算法

4.1 概述

令牌桶演演算法可以認為是漏桶演演算法的一種升級，它不但可以將流量做一步限制，還可以解決漏桶中無法彈性伸縮處理請求的問題。體現在現實中，類似服務大廳的門口設定門禁卡發放。發放是勻速的，請求較少時，令牌可以快取起來，供流量爆發時一次性批次獲取使用。而內部服務視窗不設限。

4.2 實現

package com.oldlu.limit;
import java.util.concurrent.*;
public class Token {
    public static void main(String[] args) throws InterruptedException {
        //令牌桶，號誌實現，容量為3
        final Semaphore semaphore = new Semaphore(3);
        //定時器，1s一個，勻速頒發令牌
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (semaphore.availablePermits() < 3){
                    semaphore.release();
                }
//                System.out.println("令牌數："+semaphore.availablePermits());
            }
        },1000,1000,TimeUnit.MILLISECONDS);
        //等待，等候令牌桶儲存
        Thread.sleep(5);
        //模擬洪峰5個請求，前3個迅速響應，後兩個排隊
        for (int i = 0; i < 5; i++) {
            semaphore.acquire();
            System.out.println("洪峰："+i);
        }
        //模擬日常請求，2s一個
        for (int i = 0; i < 3; i++) {
            Thread.sleep(1000);
            semaphore.acquire();
            System.out.println("日常："+i);
            Thread.sleep(1000);
        }
        //再次洪峰
        for (int i = 0; i < 5; i++) {
            semaphore.acquire();
            System.out.println("洪峰："+i);
        }
        //檢查令牌桶的數量
        for (int i = 0; i < 5; i++) {
            Thread.sleep(2000);
            System.out.println("令牌剩餘："+semaphore.availablePermits());
        }
    }
}

4.3 結果分析

注意結果出現的節奏！
洪峰0-2迅速被執行，說明桶中暫存了3個令牌，有效應對了洪峰
洪峰3，4被間隔性執行，得到了有效的限流
日常請求被勻速執行，間隔均勻
第二波洪峰來臨，和第一次一樣
請求過去後，令牌最終被均勻頒發，積累到3個後不再上升

4.4 應用

springcloud中gateway可以設定令牌桶實現限流控制，案例如下：

cloud:
    gateway:
      routes:
      ‐ id: limit_route
        uri: http://localhost:8080/test
        filters:
        ‐ name: RequestRateLimiter
          args:
           #限流的key，ipKeyResolver為spring中託管的Bean，需要擴充套件KeyResolver介面  
            key‐resolver: '#{@ipResolver}'
            #令牌桶每秒填充平均速率，相當於程式碼中的發放頻率
            redis‐rate‐limiter.replenishRate: 1
            #令牌桶總容量，相當於程式碼中，號誌的容量
            redis‐rate‐limiter.burstCapacity: 3

5 滑動視窗

5.1 概述

滑動視窗可以理解為細分之後的計數器，計數器粗暴的限定1分鐘內的存取次數，而滑動視窗限流將1分鐘拆為多個段，不但要求整個1分鐘內請求數小於上限，而且要求每個片段請求數也要小於上限。相當於將原來的計數週期做了多個片段拆分。更為精細。

5.2 實現

package com.oldlu.limit;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Window {
    //整個視窗的流量上限，超出會被限流
    final int totalMax = 5;
    //每片的流量上限，超出同樣會被拒絕，可以設定不同的值
    final int sliceMax = 5;
    //分多少片
    final int slice = 3;
    //視窗，分3段，每段1s，也就是總長度3s
    final LinkedList<Long> linkedList = new LinkedList<>();
    //計數器，每片一個key，可以使用HashMap，這裡為了控制檯保持有序性和可讀性，採用TreeMap
    Map<Long,AtomicInteger> map = new TreeMap();
    //心跳，每1s跳動1次，滑動視窗向前滑動一步，實際業務中可能需要手動控制滑動視窗的時機。
    ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
    //獲取key值，這裡即是時間戳（秒）
    private Long getKey(){
        return System.currentTimeMillis()/1000;
    }
    public Window(){
        //初始化視窗，當前時間指向的是最末端，前兩片其實是過去的2s
        Long key = getKey();
        for (int i = 0; i < slice; i++) {
            linkedList.addFirst(key‐i);
            map.put(key‐i,new AtomicInteger(0));
        }
        //啟動心跳任務，視窗根據時間，自動向前滑動，每秒1步
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                Long key = getKey();
                //隊尾新增最新的片
                linkedList.addLast(key);
                map.put(key,new AtomicInteger());
                //將最老的片移除
                map.remove(linkedList.getFirst());
                linkedList.removeFirst();
                System.out.println("step:"+key+":"+map);;
            }
        },1000,1000,TimeUnit.MILLISECONDS);
    }
    //檢查當前時間所在的片是否達到上限
    public boolean checkCurrentSlice(){
        long key = getKey();
        AtomicInteger integer = map.get(key);
        if (integer != null){
            return integer.get() < sliceMax ;
        }
        //預設允許存取
        return true;
    }
    //檢查整個視窗所有片的計數之和是否達到上限
    public boolean checkAllCount(){
        return map.values().stream().mapToInt(value ‐> value.get()).sum()  < totalMax;
    }
    //請求來臨....
    public void req(){
        Long key = getKey();
        //如果時間視窗未到達當前時間片，稍微等待一下
        //其實是一個保護措施，放置心跳對滑動視窗的推動滯後於當前請求
        while (linkedList.getLast()<key){
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //開始檢查，如果未達到上限，返回ok，計數器增加1
        //如果任意一項達到上限，拒絕請求，達到限流的目的
        //這裡是直接拒絕。現實中可能會設定緩衝池，將請求放入緩衝佇列暫存
        if (checkCurrentSlice() && checkAllCount()){
            map.get(key).incrementAndGet();
            System.out.println(key+"=ok:"+map);
        }else {
            System.out.println(key+"=reject:"+map);
        }
    }
    public static void main(String[] args) throws InterruptedException {
        Window window = new Window();
        //模擬10個離散的請求，相對之間有200ms間隔。會造成總數達到上限而被限流
        for (int i = 0; i < 10; i++) {
            Thread.sleep(200);
            window.req();
        }
        //等待一下視窗滑動，讓各個片的計數器都置零
        Thread.sleep(3000);
        //模擬突發請求，單個片的計數器達到上限而被限流
        System.out.println("‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐");
        for (int i = 0; i < 10; i++) {
        window.req();
        }
    }
}