guava 限流的兩種方式
阿新 • • 發佈:2019-01-03
版權宣告: https://blog.csdn.net/mengxpFighting/article/details/79117934
java中對於生產者消費者模型,或者小米手機營銷 1分鐘賣多少臺手機等都存在限流的思想在裡面。
關於限流 目前存在兩大類,從執行緒個數(jdk1.5 Semaphore)和RateLimiter速率(guava) Semaphore:從執行緒個數限流 RateLimiter:從速率限流 目前常見的演算法是漏桶演算法和令牌演算法 令牌桶演算法。相比漏桶演算法而言區別在於,令牌桶是會去勻速的生成令牌,拿到令牌才能夠進行處理,類似於勻速往桶裡放令牌 漏桶演算法是:生產者消費者模型,生產者往木桶裡生產資料,消費者按照定義的速度去消費資料 應用場景: 漏桶演算法:必須讀寫分流的情況下,限制讀取的速度 令牌桶演算法:必須讀寫分離的情況下,限制寫的速率或者小米手機飢餓營銷的場景 只賣1分種搶購1000 實現的方法都是一樣。RateLimiter來實現 對於多執行緒問題查詢時,很多時候可能使用的類都是原子性的,但是由於程式碼邏輯的問題,也可能發生執行緒安全問題
一、問題描述 某天A君突然發現自己的介面請求量突然漲到之前的10倍,沒多久該介面幾乎不可使用,並引發連鎖反應導致整個系統崩潰。如何應對這種情況呢?生活給了我們答案:比如老式電閘都安裝了保險絲,一旦有人使用超大功率的裝置,保險絲就會燒斷以保護各個電器不被強電流給燒壞。同理我們的介面也需要安裝上“保險絲”,以防止非預期的請求對系統壓力過大而引起的系統癱瘓,當流量過大時,可以採取拒絕或者引流等機制。 二、常用的限流演算法 常用的限流演算法有兩種:漏桶演算法和令牌桶演算法,這篇博文介紹得比較清晰(過載保護演算法淺析)。 漏桶演算法思路很簡單,請求先進入到漏桶裡,漏桶以一定的速度出水,當水請求過大會直接溢位,可以看出漏桶演算法能強行限制資料的傳輸速率。 圖1 漏桶演算法示意圖 對於很多應用場景來說,除了要求能夠限制資料的平均傳輸速率外,還要求允許某種程度的突發傳輸。這時候漏桶演算法可能就不合適了,令牌桶演算法更為適合。如圖2所示,令牌桶演算法的原理是系統會以一個恆定的速度往桶裡放入令牌,而如果請求需要被處理,則需要先從桶裡獲取一個令牌,當桶裡沒有令牌可取時,則拒絕服務。 圖2 令牌桶演算法示意圖 三、限流工具類RateLimiter google開源工具包guava提供了限流工具類RateLimiter,該類基於“令牌桶演算法”,非常方便使用。該類的介面描述請參考:RateLimiter介面描述,具體的使用請參考:RateLimiter使用實踐。
1.關於RateLimter和Semphore簡單用法
package concurrent; import com.google.common.util.concurrent.RateLimiter; import java.util.concurrent.*; import java.util.stream.IntStream; import static java.lang.Thread.currentThread; /** * ${DESCRIPTION} * 關於限流 目前存在兩大類,從執行緒個數(jdk1.5 Semaphore)和RateLimiter速率(guava) * Semaphore:從執行緒個數限流 * RateLimiter:從速率限流 目前常見的演算法是漏桶演算法和令牌演算法,下面會具體介紹 * * @author mengxp * @version 1.0 * @create 2018-01-15 22:44 **/ public class RateLimiterExample { //Guava 0.5的意思是 1秒中0.5次的操作,2秒1次的操作 從速度來限流,從每秒中能夠執行的次數來 private final static RateLimiter limiter=RateLimiter.create(0.5d); //同時只能有三個執行緒工作 Java1.5 從同時處理的執行緒個數來限流 private final static Semaphore sem=new Semaphore(3); private static void testSemaphore(){ try { sem.acquire(); System.out.println(currentThread().getName()+" is doing work..."); TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(10)); } catch (InterruptedException e) { e.printStackTrace(); }finally { sem.release(); System.out.println(currentThread().getName()+" release the semephore..other thread can get and do job"); } } public static void runTestSemaphore(){ ExecutorService service = Executors.newFixedThreadPool(10); IntStream.range(0,10).forEach((i)->{ //RateLimiterExample::testLimiter 這種寫法是建立一個執行緒 service.submit(RateLimiterExample::testSemaphore); }); } /** * Guava的RateLimiter */ private static void testLimiter(){ System.out.println(currentThread().getName()+" waiting " +limiter.acquire()); } //Guava的RateLimiter public static void runTestLimiter(){ ExecutorService service = Executors.newFixedThreadPool(10); IntStream.range(0,10).forEach((i)->{ //RateLimiterExample::testLimiter 這種寫法是建立一個執行緒 service.submit(RateLimiterExample::testLimiter); }); } public static void main(String[] args) { IntStream.range(0,10).forEach((a)-> System.out.println(a));//從0-9 //runTestLimiter(); runTestSemaphore(); } }
2.實現漏桶演算法
package concurrent.BucketAl;
import com.google.common.util.concurrent.Monitor;
import com.google.common.util.concurrent.RateLimiter;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static java.lang.Thread.currentThread;
/**
* ${DESCRIPTION}
*
* @author mengxp
* @version 1.0
* @create 2018-01-20 22:42
* 實現漏桶演算法 實現多執行緒生產者消費者模型 限流
**/
public class Bucket {
//定義桶的大小
private final ConcurrentLinkedQueue<Integer> container=new ConcurrentLinkedQueue<>();
private final static int BUCKET_LIMIT=1000;
//消費者 不論多少個執行緒,每秒最大的處理能力是1秒中執行10次
private final RateLimiter consumerRate=RateLimiter.create(10d);
//往桶裡面放資料時,確認沒有超過桶的最大的容量
private Monitor offerMonitor=new Monitor();
//從桶裡消費資料時,桶裡必須存在資料
private Monitor consumerMonitor=new Monitor();
/**
* 往桶裡面寫資料
* @param data
*/
public void submit(Integer data){
if (offerMonitor.enterIf(offerMonitor.newGuard(()->container.size()<BUCKET_LIMIT))){
try {
container.offer(data);
System.out.println(currentThread()+" submit.."+data+" container size is :["+container.size()+"]");
} finally {
offerMonitor.leave();
}
}else {
//這裡時候採用降級策略了。消費速度跟不上產生速度時,而且桶滿了,丟擲異常
//或者存入MQ DB等後續處理
throw new IllegalStateException(currentThread().getName()+"The bucket is ful..Pls latter can try...");
}
}
/**
* 從桶裡面消費資料
* @param consumer
*/
public void takeThenConsumer(Consumer<Integer> consumer){
if (consumerMonitor.enterIf(consumerMonitor.newGuard(()->!container.isEmpty()))){
try {
//不列印時 寫 consumerRate.acquire();
System.out.println(currentThread()+" waiting"+consumerRate.acquire());
Integer data = container.poll();
//container.peek() 只是去取出來不會刪掉
consumer.accept(data);
}finally {
consumerMonitor.leave();
}
}else {
//當木桶的消費完後,可以消費那些降級存入MQ或者DB裡面的資料
System.out.println("will consumer Data from MQ...");
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
2.1 漏桶演算法測試類
package concurrent.BucketAl;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
import static java.lang.Thread.currentThread;
/**
* ${DESCRIPTION}
*
* @author mengxp
* @version 1.0
* @create 2018-01-20 23:11
* 漏桶演算法測試
* 實現漏桶演算法 實現多執行緒生產者消費者模型 限流
**/
public class BuckerTest {
public static void main(String[] args) {
final Bucket bucket = new Bucket();
final AtomicInteger DATA_CREATOR = new AtomicInteger(0);
//生產執行緒 10個執行緒 每秒提交 50個數據 1/0.2s*10=50個
IntStream.range(0, 10).forEach(i -> {
new Thread(() -> {
for (; ; ) {
int data = DATA_CREATOR.incrementAndGet();
try {
bucket.submit(data);
TimeUnit.MILLISECONDS.sleep(200);
} catch (Exception e) {
//對submit時,如果桶滿了可能會丟擲異常
if (e instanceof IllegalStateException) {
System.out.println(e.getMessage());
//當滿了後,生產執行緒就休眠1分鐘
try {
TimeUnit.SECONDS.sleep(60);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
}
}
}).start();
});
//消費執行緒 採用RateLimiter每秒處理10個 綜合的比率是5:1
IntStream.range(0, 10).forEach(i -> {
new Thread(
() -> {
for (; ; ) {
bucket.takeThenConsumer(x -> {
System.out.println(currentThread()+"C.." + x);
});
}
}
).start();
});
}
}
3.令牌桶演算法
package concurrent.TokenBucket;
import com.google.common.util.concurrent.RateLimiter;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static java.lang.Thread.currentThread;
import static java.lang.Thread.interrupted;
/**
* ${DESCRIPTION}
*
* @author mengxp
* @version 1.0
* @create 2018-01-21 0:18
* 令牌桶演算法。相比漏桶演算法而言區別在於,令牌桶是會去勻速的生成令牌,拿到令牌才能夠進行處理,類似於勻速往桶裡放令牌
* 漏桶演算法是:生產者消費者模型,生產者往木桶裡生產資料,消費者按照定義的速度去消費資料
*
* 應用場景:
* 漏桶演算法:必須讀寫分流的情況下,限制讀取的速度
* 令牌桶演算法:必須讀寫分離的情況下,限制寫的速率或者小米手機飢餓營銷的場景 只賣1分種搶購1000
*
* 實現的方法都是一樣。RateLimiter來實現
* 對於多執行緒問題查詢時,很多時候可能使用的類都是原子性的,但是由於程式碼邏輯的問題,也可能發生執行緒安全問題
**/
public class TokenBuck {
//可以使用 AtomicInteger+容量 可以不用Queue實現
private AtomicInteger phoneNumbers=new AtomicInteger(0);
private RateLimiter rateLimiter=RateLimiter.create(20d);//一秒只能執行五次
//預設銷售500臺
private final static int DEFALUT_LIMIT=500;
private final int saleLimit;
public TokenBuck(int saleLimit) {
this.saleLimit = saleLimit;
}
public TokenBuck() {
this(DEFALUT_LIMIT);
}
public int buy(){
//這個check 必須放在success裡面做判斷,不然會產生執行緒安全問題(業務引起)
//原因當phoneNumbers=99 時 同時存在三個執行緒進來。雖然phoneNumbers原子性,但是也會發生。如果必須寫在這裡,在success
//裡面也需要加上double check
/* if (phoneNumbers.get()>=saleLimit){
throw new IllegalStateException("Phone has been sale "+saleLimit+" can not buy more...")
}*/
//目前設定超時時間,10秒內沒有搶到就丟擲異常
//這裡的TimeOut*Ratelimiter=總數 這裡的超時就是讓別人搶幾秒,所以設定總數也可以由這裡的超時和RateLimiter來計算
boolean success = rateLimiter.tryAcquire(10, TimeUnit.SECONDS);
if (success){
if (phoneNumbers.get()>=saleLimit){
throw new IllegalStateException("Phone has been sale "+saleLimit+" can not buy more...");
}
int phoneNo = phoneNumbers.getAndIncrement();
System.out.println(currentThread()+" user has get :["+phoneNo+"]");
return phoneNo;
}else {
//超時後 同一時間,很大的流量來強時,超時快速失敗。
throw new RuntimeException(currentThread()+"has timeOut can try again...");
}
}
}
3.1 令牌桶演算法的測試類
package concurrent.TokenBucket;
import java.util.stream.IntStream;
/**
* ${DESCRIPTION}
*
* @author mengxp
* @version 1.0
* @create 2018-01-21 0:40
**/
public class TokenBuckTest {
public static void main(String[] args) {
final TokenBuck tokenBuck=new TokenBuck(200);
IntStream.range(0,300).forEach(i->{
//目前測試時,讓一個執行緒搶一次,不用迴圈搶
//tokenBuck::buy 這種方式 產生一個Runnable
new Thread(tokenBuck::buy).start();
});
}
}
--------------------- 本文來自 大資料孟小鵬 的CSDN 部落格 ,全文地址請點選:https://blog.csdn.net/mengxpfighting/article/details/79117934?utm_source=copy
Moniter相關作用:https://www.cnblogs.com/hupengcool/p/4250903.html
Monitor類是作為ReentrantLock的一個替代,程式碼中使用 Monitor比使用ReentrantLock更不易出錯,可讀性也更強,並且也沒有顯著的效能損失,使用Monitor甚至有潛在的效能得到優化。
public abstract static class Guard:一個標識執行緒是否等待的布林條件,Guard類總是與單一的Monitor相關聯,Monitor可以在任意時間從任意佔用Monitor的執行緒檢查Guard,這樣程式碼的編寫將不在關心Guard是否被檢查的頻率。
public abstract boolean isSatisfied():Guard內部提供的抽象方法,isSatisfied(),當被關聯的Monitor被佔用時,Guard的此方法會被呼叫,該方法的實現必須取決於被關聯Monitor保護的狀態,並且狀態不可修改。
Monitor有幾個常用的方法
- enter():進入到當前Monitor,無限期阻塞,等待鎖。(這裡沒有Guard)
- enter(long time, TimeUnit unit):進入到當前Monitor,最多阻塞給定的時間,返回是否進入Monitor。
- tryEnter():如果可以的話立即進入Monitor,不阻塞,返回是否進入Monitor。
- enterWhen(Guard guard):進入當前Monitor,等待Guard的isSatisfied()為true後,繼續往下執行 ,但可能會被打斷; 為false,會阻塞。
- enterIf(Guard guard):如果Guard的isSatisfied()為true,進入當前Monitor。等待獲得鎖(這裡會等待獲取鎖),不需要等待Guard satisfied。
- tryEnterIf(Guard guard):如果Guard的isSatisfied()為true並且可以的話立即進入Monitor,不等待獲取鎖(這裡不等待獲取鎖),也不等待Guard satisfied。
- 感覺 enterWhen enterIf的用的區別就是前者無返回值,後者返回Boolean值。
- newGuard(Boolean b)為{@程式碼}監視器建立一個新的{@連結守護}