import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

//計(jì)數(shù)器 限流
public class CounterLimiter {

    //起始時(shí)間
    private static long startTime = System.currentTimeMillis();

    //時(shí)間間隔1000ms
    private static long interval = 1000;

    //每個(gè)時(shí)間間隔內(nèi)，限制數(shù)量
    private static long limit = 3;

    //累加器
    private static AtomicLong accumulator = new AtomicLong();

    /**
     * true 代表放行，請(qǐng)求可已通過
     * false 代表限制，不讓請(qǐng)求通過
     */
    public static boolean tryAcquire() {
        long nowTime = System.currentTimeMillis();
        //判斷是否在上一個(gè)時(shí)間間隔內(nèi)
        if (nowTime < startTime + interval) {
            //如果還在上個(gè)時(shí)間間隔內(nèi)
            long count = accumulator.incrementAndGet();
            if (count <= limit) {
                return true;
            } else {
                return false;
            }
        } else {
            //如果不在上一個(gè)時(shí)間間隔內(nèi)
            synchronized (CounterLimiter.class) {
                //防止重復(fù)初始化
                if (nowTime > startTime + interval) {
                    startTime = nowTime;
                    accumulator.set(0);
                }
            }
            //再次進(jìn)行判斷
            long count = accumulator.incrementAndGet();
            if (count <= limit) {
                return true;
            } else {
                return false;
            }
        }
    }


    // 測(cè)試
    public static void main(String[] args) {

        //線程池，用于多線程模擬測(cè)試
        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 被限制的次數(shù)
        AtomicInteger limited = new AtomicInteger(0);
        // 線程數(shù)
        final int threads = 2;
        // 每條線程的執(zhí)行輪數(shù)
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        boolean flag = tryAcquire();
                        if (!flag) {
                            // 被限制的次數(shù)累積
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有線程結(jié)束
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //輸出統(tǒng)計(jì)結(jié)果
        System.out.println("限制的次數(shù)為：" + limited.get() +
                ",通過的次數(shù)為：" + (threads * turns - limited.get()));
        System.out.println("限制的比例為：" + (float) limited.get() / (float) (threads * turns));
        System.out.println("運(yùn)行的時(shí)長(zhǎng)為：" + time + "s");
    }

}

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

//漏斗限流
public class LeakBucketLimiter {

    //桶的大小
    private static long capacity = 10;
    //流出速率,每秒兩個(gè)
    private static long rate = 2;
    //開始時(shí)間
    private static long startTime = System.currentTimeMillis();
    //桶中剩余的水
    private static AtomicLong water = new AtomicLong();

    /**
     * true 代表放行，請(qǐng)求可已通過
     * false 代表限制，不讓請(qǐng)求通過
     */
    public synchronized static boolean tryAcquire() {
        //如果桶的余量問0，直接放行
        if (water.get() == 0) {
            startTime = System.currentTimeMillis();
            water.set(1);
            return true;
        }
        //計(jì)算從當(dāng)前時(shí)間到開始時(shí)間流出的水，和現(xiàn)在桶中剩余的水
        //桶中剩余的水
        water.set(water.get() - (System.currentTimeMillis() - startTime) / 1000 * rate);
        //防止出現(xiàn)<0的情況
        water.set(Math.max(0, water.get()));
        //設(shè)置新的開始時(shí)間
        startTime += (System.currentTimeMillis() - startTime) / 1000 * 1000;
        //如果當(dāng)前水小于容量，表示可以放行
        if (water.get() < capacity) {
            water.incrementAndGet();
            return true;
        } else {
            return false;
        }
    }


    // 測(cè)試
    public static void main(String[] args) {

        //線程池，用于多線程模擬測(cè)試
        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 被限制的次數(shù)
        AtomicInteger limited = new AtomicInteger(0);
        // 線程數(shù)
        final int threads = 2;
        // 每條線程的執(zhí)行輪數(shù)
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        boolean flag = tryAcquire();
                        if (!flag) {
                            // 被限制的次數(shù)累積
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有線程結(jié)束
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //輸出統(tǒng)計(jì)結(jié)果
        System.out.println("限制的次數(shù)為：" + limited.get() +
                ",通過的次數(shù)為：" + (threads * turns - limited.get()));
        System.out.println("限制的比例為：" + (float) limited.get() / (float) (threads * turns));
        System.out.println("運(yùn)行的時(shí)長(zhǎng)為：" + time + "s");
    }

}

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

//令牌桶
public class TokenBucketLimiter {
    //桶的容量
    private static long capacity = 10;
    //放入令牌的速率,每秒2個(gè)
    private static long rate = 2;
    //上次放置令牌的時(shí)間
    private static long lastTime = System.currentTimeMillis();
    //桶中令牌的余量
    private static AtomicLong tokenNum = new AtomicLong();

    /**
     * true 代表放行，請(qǐng)求可已通過
     * false 代表限制，不讓請(qǐng)求通過
     */
    public synchronized static boolean tryAcquire() {
        //更新桶中剩余令牌的數(shù)量
        long now = System.currentTimeMillis();
        tokenNum.addAndGet((now - lastTime) / 1000 * rate);
        tokenNum.set(Math.min(capacity, tokenNum.get()));
        //更新時(shí)間
        lastTime += (now - lastTime) / 1000 * 1000;
        //桶中還有令牌就放行
        if (tokenNum.get() > 0) {
            tokenNum.decrementAndGet();
            return true;
        } else {
            return false;
        }
    }


    //測(cè)試
    public static void main(String[] args) {

        //線程池，用于多線程模擬測(cè)試
        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 被限制的次數(shù)
        AtomicInteger limited = new AtomicInteger(0);
        // 線程數(shù)
        final int threads = 2;
        // 每條線程的執(zhí)行輪數(shù)
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        boolean flag = tryAcquire();
                        if (!flag) {
                            // 被限制的次數(shù)累積
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有線程結(jié)束
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //輸出統(tǒng)計(jì)結(jié)果
        System.out.println("限制的次數(shù)為：" + limited.get() +
                ",通過的次數(shù)為：" + (threads * turns - limited.get()));
        System.out.println("限制的比例為：" + (float) limited.get() / (float) (threads * turns));
        System.out.println("運(yùn)行的時(shí)長(zhǎng)為：" + time + "s");
    }

}