Guava RateLimiter實現接口API限流

一、簡介

Guava提供的RateLimiter可以限制物理或邏輯資源的被訪問速率。RateLimit二的原理類似與令牌桶，它主要由許可發出的速率來定義，如果沒有額外的配置，許可證將按每秒許可證規定的固定速度分配，許可將被平滑地分發，若請求超過permitsPerSecond則RateLimiter按照每秒1/permitsPerSecond的速率釋放許可。

使用RateLimiter需要引入的jar包：

<!-- Guava是一種基於開源的Java庫,谷歌很多項目使用它的很多核心庫。這個庫是為了方便編碼,並減少編碼錯誤 -->
<dependency>
     <groupId>com.google.guava</groupId>
     <artifactId>guava</artifactId>
     <version>${guava.version}</version>
</dependency>

 

注意：

Guava 20（將於2016年初發布）將是支持Java 6甚至Java 7的最終Guava版本.Guava 21（理想情況下是2016年中期）將需要Java 8。

在版本21中，我們還將啟動一個新的分支，可能稱為guava-android。它將保持Java 6兼容性，允許它用於支持最小版本Gingerbread的Android應用程序。

 方法摘要：

參考地址：https://www.cnblogs.com/exceptioneye/p/4824394.html

返回類型	方法和描述
static RateLimiter	create(double permitsPerSecond) 根據指定的穩定吞吐率和預熱期來創建RateLimiter，這里的吞吐率是指每秒多少許可數（通常是指QPS,每秒多少查詢）
static RateLimiter	create(double permitsPerSecond,Long warmupPeriod,TimeUnti unit) 根據指定的穩定吞吐率和預熱期來創建RateLimiter，這里的吞吐率是指每秒多少許可數（通常是指QPS，每秒多少請求量），在這段預熱時間內，RateLimiter每秒分配許可數會平穩的增長直到預熱期結束是達到其最大速率。（只要存在足夠請求數來時其飽和）
double	acquire() 從RateLimiter獲取一個許可，該方法會被阻塞直到獲取到請求。
double	acquire(int permits) 從RateLimiter獲取指定數量許可，該方法會被阻塞直到獲取到請求。
double	getRate() 返回RateLimiter配置中的穩定速率，該速率單位是每秒多少許可數
void	setRate(double pemitsPerSecond) 更新RateLimiter的穩定速率，參數permitsPerSecond由構造RateLimiter的工廠方法提供。
boolean	tryAcquire() 從RateLimiter獲取許可，如果該許可可以在無延遲下的情況下立即獲取的話返回true，否則返回false。
boolean	tryAcquire(int permits) 從RateLimiter中獲取指定數量許可，如果該 許可數可以在無延遲的情況下立即獲取得到返回true，否則返回false
boolean	tryAcquire(int permits,long timeout,TimeUnit unit) 從RateLimiter中獲取指定數量許可，如果該許可可以在不超過timeout的時間內獲取得到的話返回true，如果無法在timeout時間內獲取到許可則返回false。 簡述：在指定時間(timeout)內獲取指定數量(permits)許可。
boolean	tryAcquire(long timeout,TimeUnit unit) 從RateLimiter中獲取一個許可，如果該許可可以在不超過timeout的時間內獲取得到的話返回true，如果無法在timeout時間內獲取到許可則返回false 簡述：在指定時間內獲取一個許可。

二、Demo

@Test
public void rateLimit(){ String start = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()); // 這里的1表示每秒允許處理的量為1個 RateLimiter limiter = RateLimiter.create(1.0); for (int i = 1; i <= 10; i++) { // 請求RateLimiter, 超過permits會被阻塞  limiter.acquire(); System.out.println("count => " + i); } String end = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()); System.out.println("start time:" + start); System.out.println("end time:" + end); }

 

三、自定義注解

1.標識注解：

注解：

import java.lang.annotation.*;

/**
 * 自定義限流注解（標識注解）
 *
 * @Target：
 *          表示該注解可以用於什么地方，可能的ElementType參數有：
 *          CONSTRUCTOR：構造器的聲明
 *          FIELD：域聲明（包括enum實例）
 *          LOCAL_VARIABLE：局部變量聲明
 *          METHOD：方法聲明
 *          PACKAGE：包聲明
 *          PARAMETER：參數聲明
 *          TYPE：類、接口（包括注解類型）或enum聲明
 *
 * @Retention
 *          表示需要在什么級別保存該注解信息。可選的RetentionPolicy參數包括：
 *          SOURCE：注解將被編譯器丟棄
 *          CLASS：注解在class文件中可用，但會被VM丟棄
 *          RUNTIME：VM將在運行期間保留注解，因此可以通過反射機制讀取注解的信息
 *
 * @Document
 *          將注解包含在Javadoc中
 *
 * @Inherited
 *          允許子類繼承父類中的注解
 *
 * @Date: 2019-04-09 16:31
 */ @Inherited @Documented @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) public @interface Limiting { }

 切面：

import com.alibaba.fastjson.JSON;
import com.github.aspect.util.ResponseUtil; import com.github.enums.http.HttpStatusEnum; import com.github.vo.util.ResultUtil; import com.google.common.util.concurrent.RateLimiter; import lombok.extern.slf4j.Slf4j; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.annotation.Pointcut; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.servlet.http.HttpServletResponse; /** * @Date: 2019-04-09 17:06 */ @Slf4j @Aspect @Component public class LimitingAspect { @Autowired public HttpServletResponse response; /** * 比如說，我這里設置"並發數"為5 */ private RateLimiter rateLimiter = RateLimiter.create(5.0); /** * 以注解為切入點 */ @Pointcut("@annotation(com.github.annotation.Limiting)") public void pointcut() { } @Around("pointcut()") public Object around(ProceedingJoinPoint joinPoint) { // 獲取許可。如果有返回true，否則返回false Boolean flag = rateLimiter.tryAcquire(); Object obj = null; try { if (flag) { obj = joinPoint.proceed(); } else { // 失敗返回客戶端的信息,例如：{"code":500,"msg":"服務器繁忙，請稍后再試！"} String result = JSON.toJSONString(ResultUtil.fail(HttpStatusEnum.BUSY_SERVER)); ResponseUtil.output(response, result); } } catch (Throwable e) { e.printStackTrace(); } return obj; } }

使用：

import com.github.annotation.Limiting;
import com.github.annotation.RateLimit; import com.github.vo.Result; import com.github.vo.util.ResultUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.ResponseBody; import java.util.concurrent.atomic.AtomicInteger; /** * 測試RateLimit限流（自定義注解+切面） * @Date: 2019-04-04 14:17 */ @Slf4j @Controller public class AppController { /** * AtomicInteger是一個線程安全的具有原子性的能返回int類型的對象 */ private static AtomicInteger num = new AtomicInteger(0); @Limiting @GetMapping("app") @ResponseBody public Result app(){ // num.incrementAndGet()表示先對自身進行+1操作，再返回 log.info("app() i = {}",num.incrementAndGet()); return ResultUtil.success(); } }

測壓結果：

2.帶參注解

注解：

import java.lang.annotation.*;
import java.util.concurrent.TimeUnit; /** * * 可控的限流注解 * 相對@Limiting來說高級一點 * * @Date: 2019-04-10 10:35 */ @Documented @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) public @interface RateLimit { String value() default ""; /** * 每秒放入桶中的令牌數，默認最大即不限流 * @return */ double perSecond() default Double.MAX_VALUE; /** * 獲取令牌的等待時間 默認1 * @return */ int timeOut() default 1; /** * 超時時間單位 默認：秒 * @return */ TimeUnit timeOutUnit() default TimeUnit.SECONDS; }

 切面：

import com.alibaba.fastjson.JSON;
import com.github.annotation.RateLimit; import com.github.aspect.util.ResponseUtil; import com.github.enums.http.HttpStatusEnum; import com.github.vo.util.ResultUtil; import com.google.common.util.concurrent.RateLimiter; import lombok.extern.slf4j.Slf4j; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.Signature; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.annotation.Pointcut; import org.aspectj.lang.reflect.MethodSignature; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.servlet.http.HttpServletResponse; import java.lang.reflect.Method; /** * @Date: 2019-04-10 10:40 */ @Slf4j @Aspect @Component public class RateLimitAspect {
 @Autowired private HttpServletResponse response; private RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE); /** * 定義切入點 * 兩種方式可用： * 1.通過指定包或者指定類切入 * @Pointcut("execution(public * com.github.controller.*.*(..))") * 2.通過指定注解切入 * @Pointcut("@annotation(com.github.annotation.LxRateLimit)") */ @Pointcut("@annotation(com.github.annotation.RateLimit)") public void pointcut() { } @Around("pointcut()") public Object around(ProceedingJoinPoint joinPoint) throws Throwable { log.info("攔截到了{}方法...", joinPoint.getSignature().getName()); Object obj = null; // 獲取目標方法 Signature signature = joinPoint.getSignature(); MethodSignature methodSignature = (MethodSignature)signature; Method targetMethod = methodSignature.getMethod(); if (targetMethod.isAnnotationPresent(RateLimit.class)) { // 獲取目標方法的@LxRateLimit注解 RateLimit rateLimit = targetMethod.getAnnotation(RateLimit.class); rateLimiter.setRate(rateLimit.perSecond()); // rateLimit.timeOut() = 1000 // rateLimit.timeOutUnit() = TimeUnit.MILLISECONDS 毫秒 // 判斷能否在1秒內得到令牌，如果不能則立即返回false，不會阻塞程序 if (!rateLimiter.tryAcquire(rateLimit.timeOut(), rateLimit.timeOutUnit())) { log.info("===== 接口並發量過大 ====="); ResponseUtil.output(response, JSON.toJSONString(ResultUtil.fail(HttpStatusEnum.BUSY_SERVER))); }else{ obj = joinPoint.proceed(); } } return obj; } }

使用：

import com.github.annotation.Limiting;
import com.github.annotation.RateLimit; import com.github.vo.Result; import com.github.vo.util.ResultUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.ResponseBody; import javax.servlet.http.HttpServletRequest; import java.util.concurrent.atomic.AtomicInteger; /** * 測試RateLimit限流（自定義注解+切面） * * @Date: 2019-04-04 14:17 */ @Slf4j @Controller public class AppController { /** * AtomicInteger是一個線程安全的具有原子性的能返回int類型的對象 */ private static AtomicInteger num = new AtomicInteger(0); /** * perSecond = 0.5:每秒創建0.5個 * timeOut = 1:在一秒的時間內如果有許可則成功，沒有則失敗 * * * @return */ @RateLimit(perSecond = 0.5,timeOut = 1) @GetMapping("apps") @ResponseBody public Result apps(){ log.info("app() num = {}",num.incrementAndGet()); return ResultUtil.success(); } }

測壓結果：