快捷導(dǎo)航

Java限流實現(xiàn)的幾種方法詳解

更新時間：2022年12月03日 08:32:04 作者：tcoding

這篇文章主要介紹了Java限流實現(xiàn)的幾種方法，通俗的說，限流就是限制一段時間內(nèi)，用戶訪問資源的次數(shù)，減輕服務(wù)器壓力，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

計數(shù)器

計數(shù)器限流方式比較粗暴，一次訪問就增加一次計數(shù)，在系統(tǒng)內(nèi)設(shè)置每 N 秒的訪問量，超過訪問量的訪問直接丟棄，從而實現(xiàn)限流訪問。

具體大概是以下步驟：

將時間劃分為固定的窗口大小，例如 1 s;
在窗口時間段內(nèi)，每來一個請求，對計數(shù)器加 1；
當(dāng)計數(shù)器達(dá)到設(shè)定限制后，該窗口時間內(nèi)的后續(xù)請求都將被丟棄；
該窗口時間結(jié)束后，計數(shù)器清零，從新開始計數(shù)。

這種算法的弊端

在開始的時間，訪問量被使用完后，1 s 內(nèi)會有很長時間的真空期是處于接口不可用的狀態(tài)的，同時也有可能在一秒內(nèi)出現(xiàn)兩倍的訪問量。

T窗口的前1/2時間無流量進(jìn)入，后1/2時間通過5個請求；

T+1窗口的前 1/2時間通過5個請求，后1/2時間因達(dá)到限制丟棄請求。
因此在 T的后1/2和(T+1)的前1/2時間組成的完整窗口內(nèi)，通過了10個請求。

代碼實現(xiàn)

 private final Semaphore count = new Semaphore(5);
 @PostConstruct
    public void init() {
        //初始化定時任務(wù)線程池
        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(2, t -> {
            Thread thread = new Thread(t);
            thread.setName("limit");
            return thread;
        });
        // 每10s執(zhí)行5次
        service.scheduleAtFixedRate(() -> count.release(5), 10, 10, TimeUnit.SECONDS);
  }
 	/**
     * 計數(shù)器限流
     */
    public void count() {
        try {
            count.acquire();
            System.out.println("count");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

信號量

控制并發(fā)訪問量

具體大概是以下步驟：

初始化信號量
每個請求獲取信號量，請求完釋放

代碼實現(xiàn)

	private final Semaphore flag = new Semaphore(5);
	/**
     * 信號量限流
     */
    public void flag() {
        try {
            flag.acquire();
            System.out.println("flag");
            int i = new Random().nextInt(10);
            TimeUnit.SECONDS.sleep(i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            flag.release();
        }
    }

滑動窗口

具體大概是以下步驟：

將時間劃分為細(xì)粒度的區(qū)間每個區(qū)間
維持一個計數(shù)器，每進(jìn)入一個請求則將計數(shù)器加一；
多個區(qū)間組成一個時間窗口，每流逝一個區(qū)間時間后，則拋棄最老的一個區(qū)間，納入新區(qū)間。如圖中示例的窗口 T1 變?yōu)榇翱?T2；
若當(dāng)前窗口的區(qū)間計數(shù)器總和超過設(shè)定的限制數(shù)量，則本窗口內(nèi)的后續(xù)請求都被丟棄。

代碼實現(xiàn)

  private final AtomicInteger[] window = new AtomicInteger[10];
 @PostConstruct
    public void init() {
        //初始化定時任務(wù)線程池
        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(2, t -> {
            Thread thread = new Thread(t);
            thread.setName("limit");
            return thread;
        });
        // 10個窗口，每次滑動1s
        Arrays.fill(window, new AtomicInteger(0));
        service.scheduleAtFixedRate(() -> {
            int index = (int) (System.currentTimeMillis() / 1000 % 10);
            window[index] = new AtomicInteger(0);
        }, 1, 1, TimeUnit.SECONDS);
}
 	/**
     * 滑動窗口
     */
    public void window() {
        int sum = 0;
        for (int i = 0; i < window.length; i++) {
            sum += window[i].get();
        }
        if (sum > 10) {
            return;
        }
        System.out.println("window");
        int index = (int) (System.currentTimeMillis() / 1000 % 10);
        window[index].getAndAdd(1);
    }

漏桶

具體大概是以下步驟：

初始化一個隊列，做桶
每個請求入隊列，隊列滿則阻塞
啟動定時任務(wù)，以固定的速率執(zhí)行，執(zhí)行時判讀一下入隊時間，如果延遲太久，直接丟棄（有可能客戶端已經(jīng)超時，服務(wù)端還沒有處理）

代碼實現(xiàn)

 private final BlockingQueue<Long> queue = new LinkedBlockingDeque<>(5);
  @PostConstruct
    public void init() {
        //初始化定時任務(wù)線程池
        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(2, t -> {
            Thread thread = new Thread(t);
            thread.setName("limit");
            return thread;
        });
        // 一恒定的速率執(zhí)行
        service.scheduleAtFixedRate(() -> {
            try {
                if (System.currentTimeMillis() - queue.take() > 1000L) {
                    process();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 100, 100, TimeUnit.MILLISECONDS);
}
	/**
     * 漏桶限流
     */
    public void bucket() {
        try {
            queue.put(System.currentTimeMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
  private void process() {
        System.out.println("process");
    }

令牌桶

令牌桶算法是漏斗算法的改進(jìn)版，為了處理短時間的突發(fā)流量而做了優(yōu)化，令牌桶算法主要由三部分組成：令牌流、數(shù)據(jù)流、令牌桶。

名詞釋義：

令牌桶：流通令牌的管道，用于生成的令牌的流通，放入令牌桶中。
數(shù)據(jù)流：進(jìn)入系統(tǒng)的數(shù)據(jù)流量。
令牌桶：保存令牌的區(qū)域，可以理解為一個緩沖區(qū)，令牌保存在這里用于使用。

具體大概是以下步驟：

初始化一個隊列做桶，大小為通的大小
啟動定時任務(wù)，以一定的速率往隊列中放入令牌
每個請求來臨，去隊列中獲取令牌，獲取成功正執(zhí)行，否則阻塞

代碼實現(xiàn)

private final BlockingQueue<Integer> token = new LinkedBlockingDeque<>(5);
  @PostConstruct
    public void init() {
        //初始化定時任務(wù)線程池
        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(2, t -> {
            Thread thread = new Thread(t);
            thread.setName("limit");
            return thread;
        });
        // 以恒定的速率放入令牌
        service.scheduleAtFixedRate(() -> {
            try {
                token.put(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 1, 1, TimeUnit.SECONDS);
    }
    public void token() {
        try {
            token.take();
            System.out.println("token");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

測試

  @Resource
    private LimitDemo demo;
    @Test
    public void count() throws InterruptedException {
        process(() -> demo.count());
    }
    @Test
    public void flag() throws InterruptedException {
        process(() -> demo.flag());
    }
    @Test
    public void window() throws InterruptedException {
        process(() -> demo.window());
    }
    @Test
    public void bucket() throws InterruptedException {
        process(() -> demo.bucket());
    }
    @Test
    public void token() throws InterruptedException {
        process(() -> demo.token());
    }
    private void process(Process process) throws InterruptedException {
        CompletableFuture<?>[] objects = IntStream.range(0, 10).mapToObj(i -> CompletableFuture.runAsync(() -> {
            while (true) {
                process.execute();
            }
        })).collect(Collectors.toList()).toArray(new CompletableFuture<?>[] {});
        CompletableFuture.allOf(objects);
        new CountDownLatch(1).await();
    }
    @FunctionalInterface
    public interface Process {
        void execute();
    }