public class CounterRateLimiter extends MyRateLimiter {
    /**
     * 每秒限制請(qǐng)求數(shù)
     */
    private final long permitsPerSecond;
    /**
     * 上一個(gè)窗口的開始時(shí)間
     */
    public long timestamp = System.currentTimeMillis();
    /**
     * 計(jì)數(shù)器
     */
    private int counter;

    public CounterRateLimiter(long permitsPerSecond) {
        this.permitsPerSecond = permitsPerSecond;
    }

    @Override
    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        // 窗口內(nèi)請(qǐng)求數(shù)量小于閾值，更新計(jì)數(shù)放行，否則拒絕請(qǐng)求
        if (now - timestamp < 1000) {
            if (counter < permitsPerSecond) {
                counter++;
                return true;
            } else {
                return false;
            }
        }
        // 時(shí)間窗口過期，重置計(jì)數(shù)器和時(shí)間戳
        counter = 0;
        timestamp = now;
        return true;
    }
}

public class SlidingWindowRateLimiter extends MyRateLimiter {
    /**
     * 每分鐘限制請(qǐng)求數(shù)
     */
    private final long permitsPerMinute;
    /**
     * 計(jì)數(shù)器, k-為當(dāng)前窗口的開始時(shí)間值秒，value為當(dāng)前窗口的計(jì)數(shù)
     */
    private final TreeMap<Long, Integer> counters;

    public SlidingWindowRateLimiter(long permitsPerMinute) {
        this.permitsPerMinute = permitsPerMinute;
        this.counters = new TreeMap<>();
    }

    @Override
    public synchronized boolean tryAcquire() {
        // 獲取當(dāng)前時(shí)間的所在的子窗口值； 10s一個(gè)窗口
        long currentWindowTime = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC) / 10 * 10;
        // 獲取當(dāng)前窗口的請(qǐng)求總量
        int currentWindowCount = getCurrentWindowCount(currentWindowTime);
        if (currentWindowCount >= permitsPerMinute) {
            return false;
        }
        // 計(jì)數(shù)器 + 1
        counters.merge(currentWindowTime, 1, Integer::sum);
        return true;
    }
    /**
     * 獲取當(dāng)前窗口中的所有請(qǐng)求數(shù)（并刪除所有無效的子窗口計(jì)數(shù)器）
     *
     * @param currentWindowTime 當(dāng)前子窗口時(shí)間
     * @return 當(dāng)前窗口中的計(jì)數(shù)
     */
    private int getCurrentWindowCount(long currentWindowTime) {
        // 計(jì)算出窗口的開始位置時(shí)間
        long startTime = currentWindowTime - 50;
        int result = 0;

        // 遍歷當(dāng)前存儲(chǔ)的計(jì)數(shù)器，刪除無效的子窗口計(jì)數(shù)器，并累加當(dāng)前窗口中的所有計(jì)數(shù)器之和
        Iterator<Map.Entry<Long, Integer>> iterator = counters.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Long, Integer> entry = iterator.next();
            if (entry.getKey() < startTime) {
                iterator.remove();
            } else {
                result += entry.getValue();
            }
        }
        return result;
    }
}

public class LeakyBucketRateLimiter extends MyRateLimiter {
    // 桶的容量
    private final int capacity;
    // 漏出速率
    private final int permitsPerSecond;
    // 剩余水量
    private long leftWater;
    // 上次注入時(shí)間
    private long timeStamp = System.currentTimeMillis();

    public LeakyBucketRateLimiter(int permitsPerSecond, int capacity) {
        this.capacity = capacity;
        this.permitsPerSecond = permitsPerSecond;
    }

    @Override
    public synchronized boolean tryAcquire() {
        //1. 計(jì)算剩余水量
        long now = System.currentTimeMillis();
        long timeGap = (now - timeStamp) / 1000;
        leftWater = Math.max(0, leftWater - timeGap * permitsPerSecond);
        timeStamp = now;
        
        // 如果未滿，則放行；否則限流
        if (leftWater < capacity) {
            leftWater += 1;
            return true;
        }
        return false;
    }
}

@Test
public void testLeakyBucketRateLimiter() throws InterruptedException {
    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    ExecutorService singleThread = Executors.newSingleThreadExecutor();

    LeakyBucketRateLimiter rateLimiter = new LeakyBucketRateLimiter(20, 20);
    // 存儲(chǔ)流量的隊(duì)列
    Queue<Integer> queue = new LinkedList<>();
    // 模擬請(qǐng)求  不確定速率注水
    singleThread.execute(() -> {
        int count = 0;
        while (true) {
            count++;
            boolean flag = rateLimiter.tryAcquire();
            if (flag) {
                queue.offer(count);
                System.out.println(count + "--------流量被放行--------");
            } else {
                System.out.println(count + "流量被限制");
            }
            try {
                Thread.sleep((long) (Math.random() * 1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
  
    // 模擬處理請(qǐng)求 固定速率漏水
    scheduledExecutorService.scheduleAtFixedRate(() -> {
        if (!queue.isEmpty()) {
            System.out.println(queue.poll() + "被處理");
        }
    }, 0, 100, TimeUnit.MILLISECONDS);

    // 保證主線程不會(huì)退出
    while (true) {
        Thread.sleep(10000);
    }
}

public class TokenBucketRateLimiter extends MyRateLimiter {
    /**
     * 令牌桶的容量「限流器允許的最大突發(fā)流量」
     */
    private final long capacity;
    /**
     * 令牌發(fā)放速率
     */
    private final long generatedPerSeconds;
    /**
     * 最后一個(gè)令牌發(fā)放的時(shí)間
     */
    long lastTokenTime = System.currentTimeMillis();
    /**
     * 當(dāng)前令牌數(shù)量
     */
    private long currentTokens;

    public TokenBucketRateLimiter(long generatedPerSeconds, int capacity) {
        this.generatedPerSeconds = generatedPerSeconds;
        this.capacity = capacity;
    }

    /**
     * 嘗試獲取令牌
     *
     * @return true表示獲取到令牌，放行；否則為限流
     */
    @Override
    public synchronized boolean tryAcquire() {
          /**
           * 計(jì)算令牌當(dāng)前數(shù)量
           * 請(qǐng)求時(shí)間在最后令牌是產(chǎn)生時(shí)間相差大于等于額1s（為啥時(shí)1s？因?yàn)樯闪钆频淖钚r(shí)間單位時(shí)s），則
           * 1. 重新計(jì)算令牌桶中的令牌數(shù)
           * 2. 將最后一個(gè)令牌發(fā)放時(shí)間重置為當(dāng)前時(shí)間
           */
        long now = System.currentTimeMillis();
        if (now - lastTokenTime >= 1000) {
            long newPermits = (now - lastTokenTime) / 1000 * generatedPerSeconds;
            currentTokens = Math.min(currentTokens + newPermits, capacity);
            lastTokenTime = now;
        }
        if (currentTokens > 0) {
            currentTokens--;
            return true;
        }
        return false;
    }
}

public class SlidingLogRateLimiter extends MyRateLimiter {
    /**
     * 每分鐘限制請(qǐng)求數(shù)
     */
    private static final long PERMITS_PER_MINUTE = 60;
    /**
     * 請(qǐng)求日志計(jì)數(shù)器, k-為請(qǐng)求的時(shí)間（秒），value當(dāng)前時(shí)間的請(qǐng)求數(shù)量
     */
    private final TreeMap<Long, Integer> requestLogCountMap = new TreeMap<>();

    @Override
    public synchronized boolean tryAcquire() {
        // 最小時(shí)間粒度為s
        long currentTimestamp = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC);
        // 獲取當(dāng)前窗口的請(qǐng)求總數(shù)
        int currentWindowCount = getCurrentWindowCount(currentTimestamp);
        if (currentWindowCount >= PERMITS_PER_MINUTE) {
            return false;
        }
        // 請(qǐng)求成功，將當(dāng)前請(qǐng)求日志加入到日志中
        requestLogCountMap.merge(currentTimestamp, 1, Integer::sum);
        return true;
    }

    /**
     * 統(tǒng)計(jì)當(dāng)前時(shí)間窗口內(nèi)的請(qǐng)求數(shù)
     *
     * @param currentTime 當(dāng)前時(shí)間
     * @return -
     */
    private int getCurrentWindowCount(long currentTime) {
        // 計(jì)算出窗口的開始位置時(shí)間
        long startTime = currentTime - 59;
        // 遍歷當(dāng)前存儲(chǔ)的計(jì)數(shù)器，刪除無效的子窗口計(jì)數(shù)器，并累加當(dāng)前窗口中的所有計(jì)數(shù)器之和
        return requestLogCountMap.entrySet()
                .stream()
                .filter(entry -> entry.getKey() >= startTime)
                .mapToInt(Map.Entry::getValue)
                .sum();
    }
}