建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Java實現(xiàn)系統(tǒng)限流
限流是保障系統(tǒng)高可用的方式之一，也是大廠高頻面試題，如果面試官問一句，“如何實現(xiàn)每秒鐘1000個請求的限流？”，你要是分分鐘給他寫上幾種限流方案，那豈不香哉，哈哈！話不多說，我來列幾種常用限流實現(xiàn)方式。
網(wǎng)站建設哪家好，找成都創(chuàng)新互聯(lián)公司！專注于網(wǎng)頁設計、網(wǎng)站建設、微信開發(fā)、微信小程序開發(fā)、集團企業(yè)網(wǎng)站建設等服務項目。為回饋新老客戶創(chuàng)新互聯(lián)還提供了工布江達免費建站歡迎大家使用！
1、Guava RateLimiter
Guava是Java領域很優(yōu)秀的開源項目，包含了日常開發(fā)常用的集合、String、緩存等， 其中RateLimiter是常用限流工具。
RateLimiter是基于令牌桶算法實現(xiàn)的，如果每秒10個令牌，內(nèi)部實現(xiàn)，會每100ms生產(chǎn)1個令牌。
使用Guava RateLimiter，如下：
(1) 引入pom依賴：
com.google.guava
guava
23.0
(2) 代碼：
public class GuavaRateLimiterTest {
//比如每秒生產(chǎn)10個令牌，相當于每100ms生產(chǎn)1個令牌
private RateLimiter rateLimiter = RateLimiter.create(10);
/**
* 模擬執(zhí)行業(yè)務方法
*/
public void exeBiz() {
if (rateLimiter.tryAcquire(1)) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("線程" + Thread.currentThread().getName() + "：執(zhí)行業(yè)務邏輯");
} else {
System.out.println("線程" + Thread.currentThread().getName() + "：被限流");
}
}
public static void main(String[] args) throws InterruptedException {
GuavaRateLimiterTest limiterTest = new GuavaRateLimiterTest();
Thread.sleep(500);//等待500ms，讓limiter生產(chǎn)一些令牌
//模擬瞬間生產(chǎn)100個線程請求
for (int i = 0; i < 100; i++) {
new Thread(limiterTest::exeBiz).start();
}
}
}2、滑窗計數(shù)
打個比方，某接口每秒允許100個請求，設置一個滑窗，窗口中有10個格子，每個格子占100ms，每100ms移動一次?；瑒哟翱诘母褡觿澐值脑蕉?，滑動窗口的滾動就越平滑，限流的統(tǒng)計就會越精確。
代碼如下：
/**
* 滑窗計數(shù)器
*/
public class SliderWindowRateLimiter implements Runnable {
//每秒允許的最大訪問數(shù)
private final long maxVisitPerSecond;
//將每秒時間劃分N個塊
private final int block;
//每個塊存儲的數(shù)量
private final AtomicLong[] countPerBlock;
//滑動窗口劃到了哪個塊兒，可以理解為滑動窗口的起始下標位置
private volatile int index;
//目前總的數(shù)量
private AtomicLong allCount;
/**
* 構造函數(shù)
*
* @param block，每秒鐘劃分N個窗口
* @param maxVisitPerSecond 每秒最大訪問數(shù)量
*/
public SliderWindowRateLimiter(int block, long maxVisitPerSecond) {
this.block = block;
this.maxVisitPerSecond = maxVisitPerSecond;
countPerBlock = new AtomicLong[block];
for (int i = 0; i < block; i++) {
countPerBlock[i] = new AtomicLong();
}
allCount = new AtomicLong(0);
}
/**
* 判斷是否超過最大允許數(shù)量
*
* @return
*/
public boolean isOverLimit() {
return currentQPS() > maxVisitPerSecond;
}
/**
* 獲取目前總的訪問數(shù)
*
* @return
*/
public long currentQPS() {
return allCount.get();
}
/**
* 請求訪問進來，判斷是否可以執(zhí)行業(yè)務邏輯
*/
public void visit() {
countPerBlock[index].incrementAndGet();
allCount.incrementAndGet();
if (isOverLimit()) {
System.out.println(Thread.currentThread().getName() + "被限流" + "，currentQPS：" + currentQPS() + "，index：" + index);
} else {
System.out.println(Thread.currentThread().getName() + "執(zhí)行業(yè)務邏輯" + "，currentQPS：" + currentQPS() + "，index：" + index);
}
}
/**
* 定時執(zhí)行器，
* 每N毫秒滑塊移動一次，然后再設置下新滑塊的初始化數(shù)字0，然后新的請求會落到新的滑塊上
* 同時總數(shù)減掉新滑塊上的數(shù)字，并且重置新的滑塊上的數(shù)量
*/
@Override
public void run() {
index = (index + 1) % block;
long val = countPerBlock[index].getAndSet(0);
allCount.addAndGet(-val);
}
public static void main(String[] args) {
SliderWindowRateLimiter sliderWindowRateLimiter = new SliderWindowRateLimiter(10, 100);
//固定的速率移動滑塊
ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
scheduledExecutorService.scheduleAtFixedRate(sliderWindowRateLimiter, 100, 100, TimeUnit.MILLISECONDS);
//模擬不同速度的請求
new Thread(() -> {
while (true) {
sliderWindowRateLimiter.visit();
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
//模擬不同速度的請求
new Thread(() -> {
while (true) {
sliderWindowRateLimiter.visit();
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}3、信號量
利用Semaphore，每隔固定速率，釋放Semaphore的資源。線程獲取到資源，則執(zhí)行業(yè)務代碼。
代碼如下：
public class SemaphoreOne {
private static Semaphore semaphore = new Semaphore(10);
public static void bizMethod() throws InterruptedException {
if (!semaphore.tryAcquire()) {
System.out.println(Thread.currentThread().getName() + "被拒絕");
return;
}
System.out.println(Thread.currentThread().getName() + "執(zhí)行業(yè)務邏輯");
Thread.sleep(500);//模擬處理業(yè)務邏輯需要1秒
semaphore.release();
}
public static void main(String[] args) {
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
semaphore.release(10);
System.out.println("釋放所有鎖");
}
}, 1000, 1000);
for (int i = 0; i < 10000; i++) {
try {
Thread.sleep(10);//模擬每隔10ms就有1個請求進來
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
try {
SemaphoreOne.bizMethod();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}4、令牌桶
令牌桶算法：一個存放固定容量令牌的桶，按照固定速率往桶里添加令牌，如有剩余容量則添加，沒有則放棄。如果有請求進來，則需要先從桶里獲取令牌，當桶里沒有令牌可取時，則拒絕任務。
令牌桶的優(yōu)點是：可以改變添加令牌的速率，一旦提高速率，則可以處理突發(fā)流量。
代碼如下：
public class TokenBucket {
/**
* 定義的桶
*/
public class Bucket {
//容量
int capacity;
//速率，每秒放多少
int rateCount;
//目前token個數(shù)
AtomicInteger curCount = new AtomicInteger(0);
public Bucket(int capacity, int rateCount) {
this.capacity = capacity;
this.rateCount = rateCount;
}
public void put() {
if (curCount.get() < capacity) {
System.out.println("目前數(shù)量==" + curCount.get() + ", 我還可以繼續(xù)放");
curCount.addAndGet(rateCount);
}
}
public boolean get() {
if (curCount.get() >= 1) {
curCount.decrementAndGet();
return true;
}
return false;
}
}
@Test
public void testTokenBucket() throws InterruptedException {
Bucket bucket = new Bucket(5, 2);
//固定線程，固定的速率往桶里放數(shù)據(jù)，比如每秒N個
ScheduledThreadPoolExecutor scheduledCheck = new ScheduledThreadPoolExecutor(1);
scheduledCheck.scheduleAtFixedRate(() -> {
bucket.put();
}, 0, 1, TimeUnit.SECONDS);
//先等待一會兒，讓桶里放點token
Thread.sleep(6000);
//模擬瞬間10個線程進來拿token
for (int i = 0; i < 10; i++) {
new Thread(() -> {
if (bucket.get()) {
System.out.println(Thread.currentThread() + "獲取到了資源");
} else {
System.out.println(Thread.currentThread() + "被拒絕");
}
}).start();
}
//等待，往桶里放token
Thread.sleep(3000);
//繼續(xù)瞬間10個線程進來拿token
for (int i = 0; i < 10; i++) {
new Thread(() -> {
if (bucket.get()) {
System.out.println(Thread.currentThread() + "獲取到了資源");
} else {
System.out.println(Thread.currentThread() + "被拒絕");
}
}).start();
}
}
}5、總結
本文主要介紹了幾種限流方法：Guava RateLimiter、簡單計數(shù)、滑窗計數(shù)、信號量、令牌桶，當然，限流算法還有漏桶算法、nginx限流等等。我所寫的這些方法只是個人在實際項目總使用過的，或者是早年參加阿里筆試時寫過的方式。
本文名稱：Java實現(xiàn)系統(tǒng)限流
URL網(wǎng)址：http://www.5511xx.com/article/cohodso.html
