數據結構和算法專題---4、限流算法與應用

本章我們會對限流算法做個簡單介紹，包括常用的限流算法（計數器、漏桶算法、令牌桶案發、滑動窗口）的概述、實現方式、典型場景做個說明。

什么是限流算法

限流是對系統的一種保護措施。即限制流量請求的頻率（每秒處理多少個請求）。一般來說，當請求流量超過系統的瓶頸，則丟棄掉多余的請求流量，保證系統的可用性。即要么不放進來，放進來的就保證提供服務。

計數器

概述

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

實現

package com.ls.cloud.sys.alg.limit;import com.ls.cloud.common.core.util.DateUtil;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.*;public class Counter {public static void main(String[] args) {//計數器，這里用信號量實現final Semaphore semaphore = new Semaphore(1);//定時器，到點清零ScheduledExecutorService service = Executors.newScheduledThreadPool(1);service.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {semaphore.release(1);}},3000,3000,TimeUnit.MILLISECONDS);//模擬無限請求從天而降降臨while (true) {try {//判斷計數器semaphore.acquire();} catch (InterruptedException e) {e.printStackTrace();}//如果準許響應，打印一個okDate date = new Date();SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");System.out.println("執行------------"+ dateFormat.format(date));}}
}

結果分析

執行------------2023-12-05 02:17:33
執行------------2023-12-05 02:17:36
執行------------2023-12-05 02:17:39
執行------------2023-12-05 02:17:42
執行------------2023-12-05 02:17:45

優缺點

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

典型場景

使用計數器限流的場景較少，因為它的處理邏輯不夠靈活。最常見的是登錄驗證碼倒計時，60秒接收一次，如果在限流場景使用計數器，可能導致前面100ms進入全部流程，系統可能依然會出現宕機的情況。

漏桶算法

概述

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

實現

可以基于隊列進行實現。

package com.ls.cloud.sys.alg.limit;import java.util.concurrent.*;public class Barrel {public static void main(String[] args) {//桶，用阻塞隊列實現，容量為3final LinkedBlockingQueue<Integer> que = new LinkedBlockingQueue(3);//定時器，相當于服務的窗口，2s處理一個ScheduledExecutorService service = Executors.newScheduledThreadPool(1);service.scheduleAtFixedRate(new Runnable() {@Overridepublic 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();}}}}

結果

put:1
put:2
put:3
put:4
put:5
處理：1
put:6
put:7
處理：2
put:8
put:9
處理：3
put:10
put:11
處理：5
put:12
put:13

put任務號按照順序入桶
執行任務勻速的2s一個被處理
因為桶的容量只有3，所以1-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;    }}

令牌桶

概述

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

實現

package com.ls.cloud.sys.alg.limit;import java.util.concurrent.*;public class Token {public static void main(String[] args) throws InterruptedException {//令牌桶，信號量實現，容量為3final Semaphore semaphore = new Semaphore(3);//定時器，1s一個，勻速頒發令牌ScheduledExecutorService service = Executors.newScheduledThreadPool(1);service.scheduleAtFixedRate(new Runnable() {@Overridepublic 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());}}
}

結果

洪峰：0
洪峰：1
洪峰：2
洪峰：3
洪峰：4
日常：0
日常：1
日常：2
洪峰：0
洪峰：1
洪峰：2
洪峰：3
洪峰：4
令牌剩余：2
令牌剩余：3
令牌剩余：3
令牌剩余：3
令牌剩余：3

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

典型場景

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

滑動窗口

概述

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

實現

在這里插入圖片描述

package com.ls.cloud.sys.alg.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，也就是總長度3sfinal LinkedList<Long> linkedList = new LinkedList<>();//計數器，每片一個key，可以使用HashMap，這里為了控制臺保持有序性和可讀性，采用TreeMapMap<Long,AtomicInteger> map = new TreeMap();//心跳，每1s跳動1次，滑動窗口向前滑動一步，實際業務中可能需要手動控制滑動窗口的時機。ScheduledExecutorService service = Executors.newScheduledThreadPool(1);//獲取key值，這里即是時間戳（秒）private Long getKey(){return System.currentTimeMillis()/1000;}public Window(){//初始化窗口，當前時間指向的是最末端，前兩片其實是過去的2sLong 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() {@Overridepublic 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() < totalMax;}//默認允許訪問return true;}//檢查整個窗口所有片的計數之和是否達到上限public boolean checkAllCount(){return map.values().stream().mapToInt(value -> value.get()).sum()  < sliceMax;}//請求來臨....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+"=通過:"+map);}else {System.out.println(key+"=拒絕:"+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();}}
}

結果

1701766769=通過:{1701766767=0, 1701766768=0, 1701766769=1}
1701766769=通過:{1701766767=0, 1701766768=0, 1701766769=2}
1701766769=通過:{1701766767=0, 1701766768=0, 1701766769=3}
1701766769=通過:{1701766767=0, 1701766768=0, 1701766769=4}
step:1701766770:{1701766768=0, 1701766769=4, 1701766770=0}
1701766770=通過:{1701766768=0, 1701766769=4, 1701766770=1}
1701766770=拒絕:{1701766768=0, 1701766769=4, 1701766770=1}
1701766770=拒絕:{1701766768=0, 1701766769=4, 1701766770=1}
1701766770=拒絕:{1701766768=0, 1701766769=4, 1701766770=1}
1701766770=拒絕:{1701766768=0, 1701766769=4, 1701766770=1}
step:1701766771:{1701766769=4, 1701766770=1, 1701766771=0}
1701766771=拒絕:{1701766769=4, 1701766770=1, 1701766771=0}
step:1701766772:{1701766770=1, 1701766771=0, 1701766772=0}
step:1701766773:{1701766771=0, 1701766772=0, 1701766773=0}
step:1701766774:{1701766772=0, 1701766773=0, 1701766774=0}
---------------------------
1701766774=通過:{1701766772=0, 1701766773=0, 1701766774=1}
1701766774=通過:{1701766772=0, 1701766773=0, 1701766774=2}
1701766774=通過:{1701766772=0, 1701766773=0, 1701766774=3}
1701766774=通過:{1701766772=0, 1701766773=0, 1701766774=4}
1701766774=通過:{1701766772=0, 1701766773=0, 1701766774=5}
1701766774=拒絕:{1701766772=0, 1701766773=0, 1701766774=5}
1701766774=拒絕:{1701766772=0, 1701766773=0, 1701766774=5}
1701766774=拒絕:{1701766772=0, 1701766773=0, 1701766774=5}
1701766774=拒絕:{1701766772=0, 1701766773=0, 1701766774=5}
1701766774=拒絕:{1701766772=0, 1701766773=0, 1701766774=5}
step:1701766775:{1701766773=0, 1701766774=5, 1701766775=0}
step:1701766776:{1701766774=5, 1701766775=0, 1701766776=0}
step:1701766777:{1701766775=0, 1701766776=0, 1701766777=0}
step:1701766778:{1701766776=0, 1701766777=0, 1701766778=0}