Java併發編程入門（十一）限流場景和Spring限流器實現

Java極客  |  做者  /  鏗然一葉 這是Java極客的第 39 篇原創文章

1、限流場景

限流場景通常基於硬件資源的使用負載，包括CPU，內存，IO。例如某個報表服務須要消耗大量內存，若是併發數增長就會拖慢整個應用，甚至內存溢出致使應用掛掉。

限流適用於會動態增長的資源，已經池化的資源不必定須要限流，例如數據庫鏈接池，它是已經肯定的資源，池的大小固定（即便能夠動態伸縮池大小），這種場景下並不須要經過限流來實現，只要能作到若是池內連接已經使用完，則沒法再獲取新的鏈接則可。

所以，使用限流的前提是：
1.防止資源使用過載產生不良影響。
2.使用的資源會動態增長，例如一個站點的請求。

2、Spring中實現限流

I、限流需求

1.只針對Controller限流
2.根據url請求路徑限流
3.可根據正則表達式匹配url來限流 4.可定義多個限流規則，每一個規則的最大流量不一樣

II、相關類結構

1.CurrentLimiteAspect是一個攔截器，在controller執行先後執行後攔截
2.CurrentLimiter是限流器，能夠添加限流規則，根據限流規則獲取流量通行證，釋放流量通行證；若是獲取通行證失敗則拋出異常。
3.LimiteRule是限流規則，限流規則可設置匹配url的正則表達式和最大流量值，同時獲取該規則的流量通訊證和釋放流量通訊證。
4.AcquireResult是獲取流量通訊證的結果，結果有3種：獲取成功，獲取失敗，不須要獲取。
5.Application是Spring的啓動類，簡單起見，在啓動類種添加限流規則。

III、Show me code

1.AcquireResult.java

public class AcquireResult {

    /** 獲取通行證成功 */
    public static final int ACQUIRE_SUCCESS = 0;

    /** 獲取通行證失敗 */
    public static final int ACQUIRE_FAILED = 1;

    /** 不須要獲取通行證 */
    public static final int ACQUIRE_NONEED = 2;

    /** 獲取通行證結果 */
    private int result;

    /** 可用通行證數量 */
    private int availablePermits;

    public int getResult() {
        return result;
    }

    public void setResult(int result) {
        this.result = result;
    }

    public int getAvailablePermits() {
        return availablePermits;
    }

    public void setAvailablePermits(int availablePermits) {
        this.availablePermits = availablePermits;
    }
}
複製代碼

2.LimiteRule.java

/** * @ClassName LimiteRule * @Description TODO * @Author 鏗然一葉 * @Date 2019/10/4 20:18 * @Version 1.0 * javashizhan.com **/
public class LimiteRule {

    /** 信號量 */
    private final Semaphore sema;

    /** 請求URL匹配規則 */
    private final String pattern;

    /** 最大併發數 */
    private final int maxConcurrent;

    public LimiteRule(String pattern, int maxConcurrent) {
        this.sema = new Semaphore(maxConcurrent);
        this.pattern = pattern;
        this.maxConcurrent = maxConcurrent;
    }

    /** * 獲取通行證。這裏加同步是爲了打印可用通行證數量時看起來逐個減小或者逐個增長，無此打印需求可不加synchronized關鍵字 * @param urlPath 請求Url * @return 0-獲取成功，1-沒有獲取到通行證，2-不須要獲取通行證 */
    public synchronized AcquireResult tryAcquire(String urlPath) {

        AcquireResult acquireResult = new AcquireResult();
        acquireResult.setAvailablePermits(this.sema.availablePermits());

        try {
            //Url請求匹配規則則獲取通行證
            if (Pattern.matches(pattern, urlPath)) {

                boolean acquire = this.sema.tryAcquire(50, TimeUnit.MILLISECONDS);

                if (acquire) {
                    acquireResult.setResult(AcquireResult.ACQUIRE_SUCCESS);
                    print(urlPath);
                } else {
                    acquireResult.setResult(AcquireResult.ACQUIRE_FAILED);
                }
            } else {
                acquireResult.setResult(AcquireResult.ACQUIRE_NONEED);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return acquireResult;
    }

    /** * 釋放通行證，這裏加同步是爲了打印可用通行證數量時看起來逐個減小或者逐個增長，無此打印需求可不加synchronized關鍵字 */
    public synchronized void release() {
        this.sema.release();
        print(null);
    }

    /** * 獲得最大併發數 * @return */
    public int getMaxConcurrent() {
        return this.maxConcurrent;
    }

    /** * 獲得匹配表達式 * @return */
    public String getPattern() {
        return this.pattern;
    }

    /** * 打印日誌 * @param urlPath */
    private void print(String urlPath) {
        StringBuffer buffer = new StringBuffer();
        buffer.append("Pattern: ").append(pattern).append(", ");
        if (null != urlPath) {
            buffer.append("urlPath: ").append(urlPath).append(", ");
        }
        buffer.append("Available Permits:").append(this.sema.availablePermits());
        System.out.println(buffer.toString());
    }

}
複製代碼

3.CurrentLimiter.java

/** * @ClassName CurrentLimiter * @Description TODO * @Author 鏗然一葉 * @Date 2019/10/4 20:18 * @Version 1.0 * javashizhan.com **/
public class CurrentLimiter {

    /** 本地線程變量，存儲一次請求獲取到的通行證，和其餘併發請求隔離開，在controller執行完後釋放本次請求得到的通行證 */
    private static ThreadLocal<Vector<LimiteRule>> localAcquiredLimiteRules = new ThreadLocal<Vector<LimiteRule>>();

    /** 全部限流規則 */
    private static Vector<LimiteRule> allLimiteRules = new Vector<LimiteRule>();

    /** 私有構造器，避免實例化 */
    private CurrentLimiter() {}

    /** * 添加限流規則，在spring啓動時添加，不須要加鎖，若是在運行中動態添加，須要加鎖 * @param rule */
    public static void addRule(LimiteRule rule) {
        printRule(rule);
        allLimiteRules.add(rule);
    }

    /** * 獲取流量通訊證，全部流量規則都要獲取後才能經過，若是一個不能獲取則拋出異常 * 多線程併發，須要加鎖 * @param urlPath */
    public static void tryAcquire(String urlPath) throws Exception {
        //有限流規則則處理
        if (allLimiteRules.size() > 0) {

            //能獲取到通行證的流量規則要保存下來，在Controller執行完後要釋放
            Vector<LimiteRule> acquiredLimitRules = new Vector<LimiteRule>();

            for(LimiteRule rule:allLimiteRules) {
                //獲取通行證
                AcquireResult acquireResult = rule.tryAcquire(urlPath);

                if (acquireResult.getResult() == AcquireResult.ACQUIRE_SUCCESS) {
                    acquiredLimitRules.add(rule);
                    //獲取到通行證的流量規則添加到本地線程變量
                    localAcquiredLimiteRules.set(acquiredLimitRules);

                } else if (acquireResult.getResult() == AcquireResult.ACQUIRE_FAILED) {
                    //若是獲取不到通行證則拋出異常
                    StringBuffer buffer = new StringBuffer();
                    buffer.append("The request [").append(urlPath).append("] exceeds maximum traffic limit, the limit is ").append(rule.getMaxConcurrent())
                            .append(", available permit is").append(acquireResult.getAvailablePermits()).append(".");

                    System.out.println(buffer);
                    throw new Exception(buffer.toString());

                } else {
                    StringBuffer buffer = new StringBuffer();
                    buffer.append("This path does not match the limit rule, path is [").append(urlPath)
                            .append("], pattern is [").append(rule.getPattern()).append("].");
                    System.out.println(buffer.toString());
                }
            }
        }
    }

    /** * 釋放獲取到的通行證。在controller執行完後掉調用（拋出異常也須要調用） */
    public static void release() {
        Vector<LimiteRule> acquiredLimitRules = localAcquiredLimiteRules.get();
        if (null != acquiredLimitRules && acquiredLimitRules.size() > 0) {
            acquiredLimitRules.forEach(rule->{
                rule.release();
            });
        }

        //destory本地線程變量，避免內存泄漏
        localAcquiredLimiteRules.remove();
    }

    /** * 打印限流規則信息 * @param rule */
    private static void printRule(LimiteRule rule) {
        StringBuffer buffer = new StringBuffer();
        buffer.append("Add Limit Rule, Max Concurrent: ").append(rule.getMaxConcurrent())
                .append(", Pattern: ").append(rule.getPattern());
        System.out.println(buffer.toString());
    }
}
複製代碼

4.CurrentLimiteAspect.java

/** * @ClassName CurrentLimiteAspect * @Description TODO * @Author 鏗然一葉 * @Date 2019/10/4 20:15 * @Version 1.0 * javashizhan.com **/
@Aspect
@Component
public class CurrentLimiteAspect {

    /** * 攔截controller，自行修改路徑 */
    @Pointcut("execution(* com.javashizhan.controller..*(..))")
    public void controller() { }

    @Before("controller()")
    public void controller(JoinPoint point) throws Exception {
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        //獲取通行證，urlPath的格式如：/limit
        CurrentLimiter.tryAcquire(request.getRequestURI());
    }

    /** * controller執行完後調用，即便controller拋出異常這個攔截方法也會被調用 * @param joinPoint */
    @After("controller()")
    public void after(JoinPoint joinPoint) {
        //釋放獲取到的通行證
        CurrentLimiter.release();
    }
}
複製代碼

5.Application.java

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        new SpringApplicationBuilder(Application.class).run(args);

        //添加限流規則
        LimiteRule rule = new LimiteRule("/limit", 4);
        CurrentLimiter.addRule(rule);
    }
}
複製代碼

IV、驗證

測試驗證碰到的兩個坑：
1.人工經過瀏覽器刷新請求發現controller是串行的
2.經過postman設置了併發測試也仍是串行的,即使設置了併發數,以下圖：

百度無果，只能自行寫代碼驗證了，代碼以下：

/** * @ClassName CurrentLimiteTest * @Description 驗證限流器 * @Author 鏗然一葉 * @Date 2019/10/5 0:51 * @Version 1.0 * javashizhan.com **/
public class CurrentLimiteTest {

    public static void main(String[] args) {
        final String limitUrlPath = "http://localhost:8080/limit";
        final String noLimitUrlPath = "http://localhost:8080/nolimit";

        //限流測試
        test(limitUrlPath);

        //休眠一會，等上一批線程執行完，方便查看日誌
        sleep(5000);

        //不限流測試
        test(noLimitUrlPath);

    }

    private static void test(String urlPath) {
        Thread[] requesters = new Thread[10];

        for (int i = 0; i < requesters.length; i++) {
            requesters[i] = new Thread(new Requester(urlPath));
            requesters[i].start();
        }
    }

    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class Requester implements Runnable {

    private final String urlPath;
    private final RestTemplate restTemplate = new RestTemplate();

    public Requester(String urlPath) {
        this.urlPath = urlPath;
    }

    @Override
    public void run() {
        String response = restTemplate.getForEntity(urlPath, String.class).getBody();
        System.out.println("response: " + response);
    }
}
複製代碼

輸出日誌以下：

Pattern: /limit, urlPath: /limit, Available Permits:3
Pattern: /limit, urlPath: /limit, Available Permits:2
Pattern: /limit, urlPath: /limit, Available Permits:1
Pattern: /limit, urlPath: /limit, Available Permits:0
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
Pattern: /limit, Available Permits:1
Pattern: /limit, Available Permits:2
Pattern: /limit, Available Permits:3
Pattern: /limit, Available Permits:4
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
複製代碼

能夠看到日誌輸出信息爲：
1.第1個測試url最大併發爲4，一次10個併發請求，有4個獲取通行證後，剩餘6個獲取通行證失敗。
2.獲取到通行證的4個請求在controller執行完後釋放了通行證。
3.第2個測試url沒有限制併發，10個請求均執行成功。

至此，限流器驗證成功。

注意：去掉同步鎖後（synchronized關鍵字），打印的日誌相似以下，能夠看到可用通行證數量不是遞增或者遞減的，但這並不代表邏輯不正確，這是由於信號量支持多個線程進入臨界區，在打印以前，可能已經減小了多個通行證，另外先執行的線程不必定先結束，因此看到的可用通訊證數量不是遞增也不是遞減的。信號量只能保證的是用掉一個通行證，就少一個。

Pattern: /limit, urlPath: /limit, Available Permits:2
Pattern: /limit, urlPath: /limit, Available Permits:1
Pattern: /limit, urlPath: /limit, Available Permits:0
Pattern: /limit, urlPath: /limit, Available Permits:2
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
Pattern: /limit, Available Permits:2
Pattern: /limit, Available Permits:4
Pattern: /limit, Available Permits:2
Pattern: /limit, Available Permits:3
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
複製代碼

end.

---

相關閱讀：
Java併發編程（一）知識地圖
Java併發編程（二）原子性
Java併發編程（三）可見性
Java併發編程（四）有序性
Java併發編程（五）建立線程方式概覽
Java併發編程入門（六）synchronized用法
Java併發編程入門（七）輕鬆理解wait和notify以及使用場景
Java併發編程入門（八）線程生命週期
Java併發編程入門（九）死鎖和死鎖定位
Java併發編程入門（十）鎖優化
Java併發編程入門（十二）生產者和消費者模式-代碼模板
Java併發編程入門（十三）讀寫鎖和緩存模板
Java併發編程入門（十四）CountDownLatch應用場景
Java併發編程入門（十五）CyclicBarrier應用場景
Java併發編程入門（十六）秒懂線程池差異
Java併發編程入門（十七）一圖掌握線程經常使用類和接口
Java併發編程入門（十八）再論線程安全

---

Java極客站點： javageektour.com/

<---此貼不易，左邊點贊！