快捷導(dǎo)航

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

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

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

計(jì)數(shù)器

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

具體大概是以下步驟：

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

這種算法的弊端

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

T窗口的前1/2時(shí)間無(wú)流量進(jìn)入，后1/2時(shí)間通過(guò)5個(gè)請(qǐng)求；

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

代碼實(shí)現(xiàn)

 private final Semaphore count = new Semaphore(5);
 @PostConstruct
    public void init() {
        //初始化定時(shí)任務(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);
  }
 	/**
     * 計(jì)數(shù)器限流
     */
    public void count() {
        try {
            count.acquire();
            System.out.println("count");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

信號(hào)量

控制并發(fā)訪問(wèn)量

具體大概是以下步驟：

初始化信號(hào)量
每個(gè)請(qǐng)求獲取信號(hào)量，請(qǐng)求完釋放

代碼實(shí)現(xiàn)

	private final Semaphore flag = new Semaphore(5);
	/**
     * 信號(hào)量限流
     */
    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();
        }
    }

滑動(dòng)窗口

具體大概是以下步驟：

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

代碼實(shí)現(xiàn)

  private final AtomicInteger[] window = new AtomicInteger[10];
 @PostConstruct
    public void init() {
        //初始化定時(shí)任務(wù)線程池
        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(2, t -> {
            Thread thread = new Thread(t);
            thread.setName("limit");
            return thread;
        });
        // 10個(gè)窗口，每次滑動(dòng)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);
}
 	/**
     * 滑動(dòng)窗口
     */
    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);
    }

漏桶

具體大概是以下步驟：

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

代碼實(shí)現(xiàn)

 private final BlockingQueue<Long> queue = new LinkedBlockingDeque<>(5);
  @PostConstruct
    public void init() {
        //初始化定時(shí)任務(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)版，為了處理短時(shí)間的突發(fā)流量而做了優(yōu)化，令牌桶算法主要由三部分組成：令牌流、數(shù)據(jù)流、令牌桶。

名詞釋義：

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

具體大概是以下步驟：

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

代碼實(shí)現(xiàn)

private final BlockingQueue<Integer> token = new LinkedBlockingDeque<>(5);
  @PostConstruct
    public void init() {
        //初始化定時(shí)任務(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();
        }
    }

測(cè)試

  @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();
    }