快捷導(dǎo)航

Java高并發(fā)系統(tǒng)限流算法的實現(xiàn)

更新時間：2022年05月18日 09:10:30 作者：趙廣陸

這篇文章主要介紹了Java高并發(fā)系統(tǒng)限流算法的應(yīng)用,在開發(fā)高并發(fā)系統(tǒng)時有三把利器用來保護(hù)系統(tǒng)：緩存、降級和限流，限流可以認(rèn)為服務(wù)降級的一種，限流是對系統(tǒng)的一種保護(hù)措施，需要的朋友可以參考下

1 概述

在開發(fā)高并發(fā)系統(tǒng)時有三把利器用來保護(hù)系統(tǒng)：緩存、降級和限流。限流可以認(rèn)為服務(wù)降級的一種，限流是對系統(tǒng)的一種保護(hù)措施。即限制流量請求的頻率（每秒處理多少個請求）。一般來說，當(dāng)請求流量超過系統(tǒng)的瓶頸，則丟棄掉多余的請求流量，保證系統(tǒng)的可用性。即要么不放進(jìn)來，放進(jìn)來的就保證提供服務(wù)。比如：延遲處理，拒絕處理，或者部分拒絕處理等等。

2 計數(shù)器限流

2.1 概述

計數(shù)器采用簡單的計數(shù)操作，到一段時間節(jié)點后自動清零

2.2 實現(xiàn)

package com.oldlu.limit;
import java.util.concurrent.*;
public class Counter {
    public static void main(String[] args) {
        //計數(shù)器，這里用信號量實現(xiàn)
        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);
        //模擬無數(shù)個請求從天而降
        while (true) {
            try {
                //判斷計數(shù)器
                semaphore.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //如果準(zhǔn)許響應(yīng)，打印一個ok
            System.out.println("ok");
        }
    }
}

2.3 結(jié)果分析

3個ok一組呈現(xiàn)，到下一個計數(shù)周期之前被阻斷

2.4 優(yōu)缺點

實現(xiàn)起來非常簡單。
控制力度太過于簡略，假如1s內(nèi)限制3次，那么如果3次在前100ms內(nèi)已經(jīng)用完，后面的900ms將只能處于阻塞狀態(tài)，白白浪費掉。

2.5 應(yīng)用

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

3 漏桶算法

3.1 概述

漏桶算法將請求緩存在桶中，服務(wù)流程勻速處理。超出桶容量的部分丟棄。漏桶算法主要用于保護(hù)內(nèi)部的處理業(yè)務(wù)，保障其穩(wěn)定有節(jié)奏的處理請求，但是無法根據(jù)流量的波動彈性調(diào)整響應(yīng)能力?，F(xiàn)實中，類似容納人數(shù)有限的服務(wù)大廳開啟了固定的服務(wù)窗口。

3.2 實現(xiàn)

package com.oldlu.limit;
import java.util.concurrent.*;
public class Barrel {
    public static void main(String[] args) {
        //桶，用阻塞隊列實現(xiàn)，容量為3
        final LinkedBlockingQueue<Integer> que = new LinkedBlockingQueue(3);
        //定時器，相當(dāng)于服務(wù)的窗口，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);
        //無數(shù)個請求，i 可以理解為請求的編號
        int i=0;
        while (true) {
            i++;
            try {
                System.out.println("put:"+i);
                //如果是put，會一直等待桶中有空閑位置，不會丟棄
//                que.put(i);
                //等待1s如果進(jìn)不了桶，就溢出丟棄
                que.offer(i,1000,TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

3.3 結(jié)果分析

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

3.4 優(yōu)缺點

有效的擋住了外部的請求，保護(hù)了內(nèi)部的服務(wù)不會過載
內(nèi)部服務(wù)勻速執(zhí)行，無法應(yīng)對流量洪峰，無法做到彈性處理突發(fā)任務(wù)
任務(wù)超時溢出時被丟棄?，F(xiàn)實中可能需要緩存隊列輔助保持一段時間
5）應(yīng)用
nginx中的限流是漏桶算法的典型應(yīng)用，配置案例如下：

http {
    #$binary_remote_addr 表示通過remote_addr這個標(biāo)識來做key，也就是限制同一客戶端ip地址。
#zone=one:10m 表示生成一個大小為10M，名字為one的內(nèi)存區(qū)域，用來存儲訪問的頻次信息。    
#rate=1r/s 表示允許相同標(biāo)識的客戶端每秒1次訪問    
    limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;
    server {
        location /limited/ {
        #zone=one 與上面limit_req_zone 里的name對應(yīng)。
#burst=5 緩沖區(qū)，超過了訪問頻次限制的請求可以先放到這個緩沖區(qū)內(nèi)，類似代碼中的隊列長度。        
#nodelay 如果設(shè)置，超過訪問頻次而且緩沖區(qū)也滿了的時候就會直接返回503，如果沒有設(shè)置，則所有請求
會等待排隊，類似代碼中的put還是offer。  
       
        limit_req zone=one burst=5 nodelay;
    }
}

4 令牌桶算法

4.1 概述

令牌桶算法可以認(rèn)為是漏桶算法的一種升級，它不但可以將流量做一步限制，還可以解決漏桶中無法彈性伸縮處理請求的問題。體現(xiàn)在現(xiàn)實中，類似服務(wù)大廳的門口設(shè)置門禁卡發(fā)放。發(fā)放是勻速的，請求較少時，令牌可以緩存起來，供流量爆發(fā)時一次性批量獲取使用。而內(nèi)部服務(wù)窗口不設(shè)限。

4.2 實現(xiàn)

package com.oldlu.limit;
import java.util.concurrent.*;
public class Token {
    public static void main(String[] args) throws InterruptedException {
        //令牌桶，信號量實現(xiàn)，容量為3
        final Semaphore semaphore = new Semaphore(3);
        //定時器，1s一個，勻速頒發(fā)令牌
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (semaphore.availablePermits() < 3){
                    semaphore.release();
                }
//                System.out.println("令牌數(shù)："+semaphore.availablePermits());
            }
        },1000,1000,TimeUnit.MILLISECONDS);
        //等待，等候令牌桶儲存
        Thread.sleep(5);
        //模擬洪峰5個請求，前3個迅速響應(yīng)，后兩個排隊
        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);
        }
        //檢查令牌桶的數(shù)量
        for (int i = 0; i < 5; i++) {
            Thread.sleep(2000);
            System.out.println("令牌剩余："+semaphore.availablePermits());
        }
    }
}

4.3 結(jié)果分析

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

4.4 應(yīng)用

springcloud中g(shù)ateway可以配置令牌桶實現(xiàn)限流控制，案例如下：

cloud:
    gateway:
      routes:
      ‐ id: limit_route
        uri: http://localhost:8080/test
        filters:
        ‐ name: RequestRateLimiter
          args:
           #限流的key，ipKeyResolver為spring中托管的Bean，需要擴(kuò)展KeyResolver接口  
            key‐resolver: '#{@ipResolver}'
            #令牌桶每秒填充平均速率，相當(dāng)于代碼中的發(fā)放頻率
            redis‐rate‐limiter.replenishRate: 1
            #令牌桶總?cè)萘浚喈?dāng)于代碼中，信號量的容量
            redis‐rate‐limiter.burstCapacity: 3

5 滑動窗口

5.1 概述

滑動窗口可以理解為細(xì)分之后的計數(shù)器，計數(shù)器粗暴的限定1分鐘內(nèi)的訪問次數(shù)，而滑動窗口限流將1分鐘拆為多個段，不但要求整個1分鐘內(nèi)請求數(shù)小于上限，而且要求每個片段請求數(shù)也要小于上限。相當(dāng)于將原來的計數(shù)周期做了多個片段拆分。更為精細(xì)。

5.2 實現(xiàn)

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;
    //每片的流量上限，超出同樣會被拒絕，可以設(shè)置不同的值
    final int sliceMax = 5;
    //分多少片
    final int slice = 3;
    //窗口，分3段，每段1s，也就是總長度3s
    final LinkedList<Long> linkedList = new LinkedList<>();
    //計數(shù)器，每片一個key，可以使用HashMap，這里為了控制臺保持有序性和可讀性，采用TreeMap
    Map<Long,AtomicInteger> map = new TreeMap();
    //心跳，每1s跳動1次，滑動窗口向前滑動一步，實際業(yè)務(wù)中可能需要手動控制滑動窗口的時機(jī)。
    ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
    //獲取key值，這里即是時間戳（秒）
    private Long getKey(){
        return System.currentTimeMillis()/1000;
    }
    public Window(){
        //初始化窗口，當(dāng)前時間指向的是最末端，前兩片其實是過去的2s
        Long key = getKey();
        for (int i = 0; i < slice; i++) {
            linkedList.addFirst(key‐i);
            map.put(key‐i,new AtomicInteger(0));
        }
        //啟動心跳任務(wù)，窗口根據(jù)時間，自動向前滑動，每秒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);
    }
    //檢查當(dāng)前時間所在的片是否達(dá)到上限
    public boolean checkCurrentSlice(){
        long key = getKey();
        AtomicInteger integer = map.get(key);
        if (integer != null){
            return integer.get() < sliceMax ;
        }
        //默認(rèn)允許訪問
        return true;
    }
    //檢查整個窗口所有片的計數(shù)之和是否達(dá)到上限
    public boolean checkAllCount(){
        return map.values().stream().mapToInt(value ‐> value.get()).sum()  < totalMax;
    }
    //請求來臨....
    public void req(){
        Long key = getKey();
        //如果時間窗口未到達(dá)當(dāng)前時間片，稍微等待一下
        //其實是一個保護(hù)措施，放置心跳對滑動窗口的推動滯后于當(dāng)前請求
        while (linkedList.getLast()<key){
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //開始檢查，如果未達(dá)到上限，返回ok，計數(shù)器增加1
        //如果任意一項達(dá)到上限，拒絕請求，達(dá)到限流的目的
        //這里是直接拒絕?，F(xiàn)實中可能會設(shè)置緩沖池，將請求放入緩沖隊列暫存
        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間隔。會造成總數(shù)達(dá)到上限而被限流
        for (int i = 0; i < 10; i++) {
            Thread.sleep(200);
            window.req();
        }
        //等待一下窗口滑動，讓各個片的計數(shù)器都置零
        Thread.sleep(3000);
        //模擬突發(fā)請求，單個片的計數(shù)器達(dá)到上限而被限流
        System.out.println("‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐");
        for (int i = 0; i < 10; i++) {
        window.req();
        }
    }
}