java併發系列 - 第29天：高併發中常見的限流方式
阿新 • • 發佈：2019-08-20
這是java高併發系列第29篇。
環境：jdk1.8。
本文內容
- 介紹常見的限流演算法
- 通過控制最大併發數來進行限流
- 通過漏桶演算法來進行限流
- 通過令牌桶演算法來進行限流
- 限流工具類RateLimiter
常見的限流的場景
- 秒殺活動，數量有限，訪問量巨大，為了防止系統宕機，需要做限流處理
- 國慶期間，一般的旅遊景點人口太多，採用排隊方式做限流處理
- 醫院看病通過發放排隊號的方式來做限流處理。
常見的限流演算法
- 通過控制最大併發數來進行限流
- 使用漏桶演算法來進行限流
- 使用令牌桶演算法來進行限流
通過控制最大併發數來進行限流
以秒殺業務為例，10個iphone，100萬人搶購，100萬人同時發起請求，最終能夠搶到的人也就是前面幾個人，後面的基本上都沒有希望了，那麼我們可以通過控制併發數來實現，比如併發數控制在10個，其他超過併發數的請求全部拒絕，提示：秒殺失敗，請稍後重試。
併發控制的，通俗解釋：一大波人去商場購物，必須經過一個門口，門口有個門衛，兜裡面有指定數量的門禁卡，來的人先去門衛那邊拿取門禁卡，拿到卡的人才可以刷卡進入商場，拿不到的可以繼續等待。進去的人出來之後會把卡歸還給門衛，門衛可以把歸還來的卡繼續發放給其他排隊的顧客使用。
JUC中提供了這樣的工具類：Semaphore，示例程式碼：
package com.itsoku.chat29; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * 跟著阿里p7學併發，微信公眾號：javacode2018 */ public class Demo1 { static Semaphore semaphore = new Semaphore(5); public static void main(String[] args) { for (int i = 0; i < 20; i++) { new Thread(() -> { boolean flag = false; try { flag = semaphore.tryAcquire(100, TimeUnit.MICROSECONDS); if (flag) { //休眠2秒，模擬下單操作 System.out.println(Thread.currentThread() + "，嘗試下單中。。。。。"); TimeUnit.SECONDS.sleep(2); } else { System.out.println(Thread.currentThread() + "，秒殺失敗，請稍微重試！"); } } catch (InterruptedException e) { e.printStackTrace(); } finally { if (flag) { semaphore.release(); } } }).start(); } } }
輸出：
Thread[Thread-10,5,main]，嘗試下單中。。。。。 Thread[Thread-8,5,main]，嘗試下單中。。。。。 Thread[Thread-9,5,main]，嘗試下單中。。。。。 Thread[Thread-12,5,main]，嘗試下單中。。。。。 Thread[Thread-11,5,main]，嘗試下單中。。。。。 Thread[Thread-2,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-1,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-18,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-16,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-0,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-3,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-14,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-6,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-13,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-17,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-7,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-19,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-15,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-4,5,main]，秒殺失敗，請稍微重試！ Thread[Thread-5,5,main]，秒殺失敗，請稍微重試！
關於Semaphore的使用，可以移步：JUC中的Semaphore（訊號量）
使用漏桶演算法來進行限流
國慶期間比較火爆的景點，人流量巨大，一般入口處會有限流的彎道，讓遊客進去進行排隊，排在前面的人，每隔一段時間會放一撥進入景區。排隊人數超過了指定的限制，後面再來的人會被告知今天已經遊客量已經達到峰值，會被拒絕排隊，讓其明天或者以後再來，這種玩法採用漏桶限流的方式。
漏桶演算法思路很簡單，水（請求）先進入到漏桶裡，漏桶以一定的速度出水，當水流入速度過大會直接溢位，可以看出漏桶演算法能強行限制資料的傳輸速率。
漏桶演算法示意圖：
簡陋版的實現，程式碼如下：
package com.itsoku.chat29;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
/**
* 跟著阿里p7學併發，微信公眾號：javacode2018
*/
public class Demo2 {
public static class BucketLimit {
static AtomicInteger threadNum = new AtomicInteger(1);
//容量
private int capcity;
//流速
private int flowRate;
//流速時間單位
private TimeUnit flowRateUnit;
private BlockingQueue<Node> queue;
//漏桶流出的任務時間間隔（納秒）
private long flowRateNanosTime;
public BucketLimit(int capcity, int flowRate, TimeUnit flowRateUnit) {
this.capcity = capcity;
this.flowRate = flowRate;
this.flowRateUnit = flowRateUnit;
this.bucketThreadWork();
}
//漏桶執行緒
public void bucketThreadWork() {
this.queue = new ArrayBlockingQueue<Node>(capcity);
//漏桶流出的任務時間間隔（納秒）
this.flowRateNanosTime = flowRateUnit.toNanos(1) / flowRate;
Thread thread = new Thread(this::bucketWork);
thread.setName("漏桶執行緒-" + threadNum.getAndIncrement());
thread.start();
}
//漏桶執行緒開始工作
public void bucketWork() {
while (true) {
Node node = this.queue.poll();
if (Objects.nonNull(node)) {
//喚醒任務執行緒
LockSupport.unpark(node.thread);
}
//休眠flowRateNanosTime
LockSupport.parkNanos(this.flowRateNanosTime);
}
}
//返回一個漏桶
public static BucketLimit build(int capcity, int flowRate, TimeUnit flowRateUnit) {
if (capcity < 0 || flowRate < 0) {
throw new IllegalArgumentException("capcity、flowRate必須大於0！");
}
return new BucketLimit(capcity, flowRate, flowRateUnit);
}
//當前執行緒加入漏桶，返回false，表示漏桶已滿；true：表示被漏桶限流成功，可以繼續處理任務
public boolean acquire() {
Thread thread = Thread.currentThread();
Node node = new Node(thread);
if (this.queue.offer(node)) {
LockSupport.park();
return true;
}
return false;
}
//漏桶中存放的元素
class Node {
private Thread thread;
public Node(Thread thread) {
this.thread = thread;
}
}
}
public static void main(String[] args) {
BucketLimit bucketLimit = BucketLimit.build(10, 60, TimeUnit.MINUTES);
for (int i = 0; i < 15; i++) {
new Thread(() -> {
boolean acquire = bucketLimit.acquire();
System.out.println(System.currentTimeMillis() + " " + acquire);
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}程式碼中BucketLimit.build(10, 60, TimeUnit.MINUTES);建立了一個容量為10，流水為60/分鐘的漏桶。
程式碼中用到的技術有：
- BlockingQueue阻塞佇列
- JUC中的LockSupport工具類，必備技能
使用令牌桶演算法來進行限流
令牌桶演算法的原理是系統以恆定的速率產生令牌，然後把令牌放到令牌桶中，令牌桶有一個容量，當令牌桶滿了的時候，再向其中放令牌，那麼多餘的令牌會被丟棄；當想要處理一個請求的時候，需要從令牌桶中取出一個令牌，如果此時令牌桶中沒有令牌，那麼則拒絕該請求。從原理上看，令牌桶演算法和漏桶演算法是相反的，一個“進水”，一個是“漏水”。這種演算法可以應對突發程度的請求，因此比漏桶演算法好。
令牌桶演算法示意圖：
有興趣的可以自己去實現一個。
限流工具類RateLimiter
Google開源工具包Guava提供了限流工具類RateLimiter，可以非常方便的控制系統每秒吞吐量，示例程式碼如下：
package com.itsoku.chat29;
import com.google.common.util.concurrent.RateLimiter;
import java.util.Calendar;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
/**
* 跟著阿里p7學併發，微信公眾號：javacode2018
*/
public class Demo3 {
public static void main(String[] args) throws InterruptedException {
RateLimiter rateLimiter = RateLimiter.create(5);//設定QPS為5
for (int i = 0; i < 10; i++) {
rateLimiter.acquire();
System.out.println(System.currentTimeMillis());
}
System.out.println("----------");
//可以隨時調整速率，我們將qps調整為10
rateLimiter.setRate(10);
for (int i = 0; i < 10; i++) {
rateLimiter.acquire();
System.out.println(System.currentTimeMillis());
}
}
}輸出：
1566284028725
1566284028922
1566284029121
1566284029322
1566284029522
1566284029721
1566284029921
1566284030122
1566284030322
1566284030522
----------
1566284030722
1566284030822
1566284030921
1566284031022
1566284031121
1566284031221
1566284031321
1566284031422
1566284031522
1566284031622程式碼中RateLimiter.create(5)建立QPS為5的限流物件，後面又呼叫rateLimiter.setRate(10);將速率設為10，輸出中分2段，第一段每次輸出相隔200毫秒，第二段每次輸出相隔100毫秒，可以非常精準的控制系統的QPS。
上面介紹的這些，業務中可能會用到，也可以用來應對面試。
java高併發系列目錄
- 第1天:必須知道的幾個概念
- 第2天:併發級別
- 第3天:有關並行的兩個重要定律
- 第4天:JMM相關的一些概念
- 第5天:深入理解程序和執行緒
- 第6天:執行緒的基本操作
- 第7天:volatile與Java記憶體模型
- 第8天:執行緒組
- 第9天：使用者執行緒和守護執行緒
- 第10天:執行緒安全和synchronized關鍵字
- 第11天:執行緒中斷的幾種方式
- 第12天JUC:ReentrantLock重入鎖
- 第13天:JUC中的Condition物件
- 第14天:JUC中的LockSupport工具類，必備技能
- 第15天：JUC中的Semaphore（訊號量）
- 第16天：JUC中等待多執行緒完成的工具類CountDownLatch，必備技能
- 第17天：JUC中的迴圈柵欄CyclicBarrier的6種使用場景
- 第18天：JAVA執行緒池，這一篇就夠了
- 第19天：JUC中的Executor框架詳解1
- 第20天：JUC中的Executor框架詳解2
- 第21天：java中的CAS，你需要知道的東西
- 第22天：JUC底層工具類Unsafe，高手必須要了解
- 第23天：JUC中原子類，一篇就夠了
- 第24天：ThreadLocal、InheritableThreadLocal（通俗易懂）
- 第25天：掌握JUC中的阻塞佇列
- 第26篇：學會使用JUC中常見的集合，常看看！
- 第27天：實戰篇，介面效能提升幾倍原來這麼簡單
- 第28天：實戰篇，微服務日誌的傷痛，一併幫你解決掉
java高併發系列連載中，總計估計會有四五十篇文章。
阿里p7一起學併發，公眾號：路人甲java，每天獲取最新文章！
