Java高並發系統限流算法的實現
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();
}
}
}
5.3 結果分析
模擬零零散散的請求，會造成每個片裡均有計數，總數達到上限後，不再響應，限流生效：
再模擬突發的流量請求，會造成單片流量計數達到上限，不再響應而被限流
5.4 應用
滑動窗口算法，在tcp協議發包過程中被使用。在web現實場景中，可以將流量控制做更細化處理，解決計數器模型控制力度太粗暴的問題。
到此這篇關於Java高並發系統限流算法的應用的文章就介紹到這瞭,更多相關Java高並發限流算法內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- Java並發編程之工具類Semaphore的使用
- java並發包中CountDownLatch和線程池的使用詳解
- 徹底搞懂Java多線程(三)
- Java並發編程之Executors類詳解
- Java使用Semaphore對單接口進行限流