Java における一般的な電流制限アルゴリズムは何ですか?

01 固定ウィンドウ

固定ウィンドウは、固定ウィンドウ (カウンタ アルゴリズム、固定ウィンドウとも呼ばれる) 電流制限アルゴリズムとも呼ばれ、最も単純な電流制限アルゴリズムです。単位時間: 単位時間内の最大訪問数。

1 分あたりのリクエスト数が 60 以下に制限されていると仮定して、カウンタを設定します。リクエストが到着したときに、カウンタがしきい値に達するとリクエストは拒否され、それ以外の場合はカウンタが増分されます。 1 ずつ増加し、カウンターは 1 分ごとに 0 にリセットされます。コードは次のように実装されます:

public class CounterRateLimiter extends MyRateLimiter {
    /**
     * 每秒限制请求数
     */
    private final long permitsPerSecond;
    /**
     * 上一个窗口的开始时间
     */
    public long timestamp = System.currentTimeMillis();
    /**
     * 计数器
     */
    private int counter;

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

    @Override
    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        // 窗口内请求数量小于阈值，更新计数放行，否则拒绝请求
        if (now - timestamp < 1000) {
            if (counter < permitsPerSecond) {
                counter++;
                return true;
            } else {
                return false;
            }
        }
        // 时间窗口过期，重置计数器和时间戳
        counter = 0;
        timestamp = now;
        return true;
    }
}

固定ウィンドウの最大の利点は、実装が簡単で、メモリ フットプリントが小さいため、タイム ウィンドウにカウントを保存するだけで済みます。古いリクエストの蓄積により新しいリクエストが枯渇してしまうことがないよう、より最新のリクエストが処理されるようにします。もちろん、重大な問題にも直面しており、2 つのウィンドウが交わるときの瞬間的なトラフィックは 2n になる可能性があります。

02 スライディングウィンドウ

瞬間的な渋滞を防ぐため、固定ウィンドウをさらに複数のグリッドに分割し、ウィンドウサイズを固定するのではなく、小さなグリッドを後ろに移動するたびに、これはスライディングウィンドウ（スライディングウィンドウ）です。

たとえば、1 分を 10 秒の 6 つのセルに分割できます。各セルにはカウンターが保持され、ウィンドウは一度に 1 セルずつ前にスライドします。リクエストが到着すると、ウィンドウ内のすべてのセルのカウントの合計がしきい値を超えない限り、リクエストを解放できます。 TCP プロトコルでのデータ パケットの送信でも、フロー制御にスライディング ウィンドウが使用されます。

実装は次のとおりです:

public class SlidingWindowRateLimiter extends MyRateLimiter {
    /**
     * 每分钟限制请求数
     */
    private final long permitsPerMinute;
    /**
     * 计数器, k-为当前窗口的开始时间值秒，value为当前窗口的计数
     */
    private final TreeMap<Long, Integer> counters;

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

    @Override
    public synchronized boolean tryAcquire() {
        // 获取当前时间的所在的子窗口值； 10s一个窗口
        long currentWindowTime = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC) / 10 * 10;
        // 获取当前窗口的请求总量
        int currentWindowCount = getCurrentWindowCount(currentWindowTime);
        if (currentWindowCount >= permitsPerMinute) {
            return false;
        }
        // 计数器 + 1
        counters.merge(currentWindowTime, 1, Integer::sum);
        return true;
    }
    /**
     * 获取当前窗口中的所有请求数（并删除所有无效的子窗口计数器）
     *
     * @param currentWindowTime 当前子窗口时间
     * @return 当前窗口中的计数
     */
    private int getCurrentWindowCount(long currentWindowTime) {
        // 计算出窗口的开始位置时间
        long startTime = currentWindowTime - 50;
        int result = 0;

        // 遍历当前存储的计数器，删除无效的子窗口计数器，并累加当前窗口中的所有计数器之和
        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;
    }
}

スライディング ウィンドウは、カウンターの瞬間的なトラフィック ピークの問題を解決します。実際、カウンター アルゴリズムもスライディング ウィンドウの一種ですが、ウィンドウより細かい単位に分割されていません。カウンタを比較すると、ウィンドウ分割の粒度が細かいほど、フロー制御がより正確かつ厳密になることがわかります。

ただし、ウィンドウ内のトラフィックがしきい値に達すると、トラフィックは即座に遮断されます。実際のアプリケーションでは、多くの場合、必要なフロー制限効果はトラフィックを一度に遮断することではなく、トラフィックがシステムにスムーズに入るようにします。

03 リーキーバケットアルゴリズム

フローをよりスムーズに制限するにはどうすればよいですか? Leaky Bucket アルゴリズムを見てみましょう。リクエストは水のように任意の速度で Leaky Bucket に注入され、バケットは一定の速度で水を漏洩します。注入速度が漏洩速度よりも大きくなり続けると、Leaky Bucket は漏洩します。いっぱいになると、この時点で新しく受信したリクエストは破棄されます。電流制限とシェーピングは、リーキー バケット アルゴリズムの 2 つの中心的な機能です。

実装は次のとおりです:

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

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

    @Override
    public synchronized boolean tryAcquire() {
        //1. 计算剩余水量
        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;
    }
}

これはリーキー バケット アルゴリズムの完全な実装ではありません。上記のコードは、トラフィックが放棄されるかどうかのみを検証します。つまり、tryAcquire は true を返します。リーキーバケットがいっぱいではないことを示します。それ以外の場合は、リーキーバケットがいっぱいでリクエストが破棄されることを意味します。

一定のレートでトラフィックをリークしたい場合は、通常、FIFO キューを使用して実装する必要があります。tryAcquire が true を返すと、リクエストはキューに入れられ、一定の間隔でキューから取り出されます。処理する周波数。サンプル コードは次のとおりです。

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

    LeakyBucketRateLimiter rateLimiter = new LeakyBucketRateLimiter(20, 20);
    // 存储流量的队列
    Queue<Integer> queue = new LinkedList<>();
    // 模拟请求  不确定速率注水
    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();
            }
        }
    });
  
    // 模拟处理请求 固定速率漏水
    scheduledExecutorService.scheduleAtFixedRate(() -> {
        if (!queue.isEmpty()) {
            System.out.println(queue.poll() + "被处理");
        }
    }, 0, 100, TimeUnit.MILLISECONDS);

    // 保证主线程不会退出
    while (true) {
        Thread.sleep(10000);
    }
}

リーキー バケット アルゴリズムの目的は、主にバースト トラフィックを平滑化し、ネットワーク内のバースト トラフィックがスムーズで安定したトラフィックに統合されるようにするメカニズムを提供することです。

ただし、リーキー バケットはトラフィックを厳密に制御しすぎるため、一部のシナリオではシステム リソースを完全に使用できません。これは、たとえダウンストリームが特定の瞬間により多くのトラフィックを処理できたとしても、リーキー バケットのリーク レートは固定されているためです。バケットではバースト トラフィックの通過も許可されません。

04トークン バケット

トラフィック レートを制限しながらバースト トラフィックを許可するにはどうすればよいですか?トークンバケットアルゴリズムについて学びましょう！トークン バケット アルゴリズムは、一定のレートでトークン バケットにトークンを送信します。リクエストが到着すると、トークン バケットからトークンを取得しようとします。トークンを取得した場合のみ解放でき、そうでない場合は拒否されます。

トークン バケットには次の特性があります:

1. トークンは一定のレートで発行されます。現在の制限レートが v/s であると仮定すると、1 回ごとに 1 つのトークンが発行されることを意味します。 /v 秒

2. トークン バケットの容量が b であると仮定します。トークン バケットがいっぱいの場合、新しいトークンは破棄されます

3. リクエストが現在のトークンを渡すための前提条件リミッターはトークン バケットです。

にトークンがあります。トークン バケット アルゴリズムには、注目に値する 2 つのパラメーター、つまり現在の制限レート v/s とトークン バケットの容量 b があります。レート a は現在の制限レートを表します。レート、b はバーストの略語で、電流リミッタによって許容される最大バースト トラフィックを示します。

例: b=10。トークン バケットがいっぱいの場合、利用可能なトークンは 10 個あります。このとき、同時に 10 個のリクエストがフロー リミッターを通過できます (ある程度のトラフィックこれら 10 個のリクエスト トークンが即座に消費されると、後続のトラフィックはレート r でのみフロー リミッターを通過できます。

実装は次のとおりです:

public class TokenBucketRateLimiter extends MyRateLimiter {
    /**
     * 令牌桶的容量「限流器允许的最大突发流量」
     */
    private final long capacity;
    /**
     * 令牌发放速率
     */
    private final long generatedPerSeconds;
    /**
     * 最后一个令牌发放的时间
     */
    long lastTokenTime = System.currentTimeMillis();
    /**
     * 当前令牌数量
     */
    private long currentTokens;

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

    /**
     * 尝试获取令牌
     *
     * @return true表示获取到令牌，放行；否则为限流
     */
    @Override
    public synchronized boolean tryAcquire() {
          /**
           * 计算令牌当前数量
           * 请求时间在最后令牌是产生时间相差大于等于额1s（为啥时1s？因为生成令牌的最小时间单位时s），则
           * 1. 重新计算令牌桶中的令牌数
           * 2. 将最后一个令牌发放时间重置为当前时间
           */
        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;
    }
}

知っておく必要があるのは、実装を考えるのが非常に簡単なのはプロデューサー/コンシューマー パターンであるということです。プロデューサー スレッドを使用して定期的にトークンを現在のリミッターを渡すスレッドはコンシューマ スレッドとして機能し、ブロッキング キューからトークンを取得した場合にのみ、現在のリミッターを渡すことが許可されます。

スレッドのスケジューリングが不確実であるため、同時実行性が高いシナリオではタイマー エラーが非常に大きくなり、同時にタイマー自体がスレッドのスケジューリングを作成するため、システムのパフォーマンスにも影響します。

05 スライディング ログ

スライディング ログは比較的「不人気」ですが、非常に使いやすい電流制限アルゴリズムです。スライディング ログ レート制限アルゴリズムでは、リクエストのタイムスタンプを記録する必要があり、これは通常、順序付けされたコレクションを使用して保存され、単一の順序付けされたコレクション内の期間内のすべてのユーザーのリクエストを追跡できます。

假设我们要限制给定T时间内的请求不超过N，我们只需要存储最近T时间之内的请求日志，每当请求到来时判断最近T时间内的请求总数是否超过阈值。

实现如下：

public class SlidingLogRateLimiter extends MyRateLimiter {
    /**
     * 每分钟限制请求数
     */
    private static final long PERMITS_PER_MINUTE = 60;
    /**
     * 请求日志计数器, k-为请求的时间（秒），value当前时间的请求数量
     */
    private final TreeMap<Long, Integer> requestLogCountMap = new TreeMap<>();

    @Override
    public synchronized boolean tryAcquire() {
        // 最小时间粒度为s
        long currentTimestamp = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC);
        // 获取当前窗口的请求总数
        int currentWindowCount = getCurrentWindowCount(currentTimestamp);
        if (currentWindowCount >= PERMITS_PER_MINUTE) {
            return false;
        }
        // 请求成功，将当前请求日志加入到日志中
        requestLogCountMap.merge(currentTimestamp, 1, Integer::sum);
        return true;
    }

    /**
     * 统计当前时间窗口内的请求数
     *
     * @param currentTime 当前时间
     * @return -
     */
    private int getCurrentWindowCount(long currentTime) {
        // 计算出窗口的开始位置时间
        long startTime = currentTime - 59;
        // 遍历当前存储的计数器，删除无效的子窗口计数器，并累加当前窗口中的所有计数器之和
        return requestLogCountMap.entrySet()
                .stream()
                .filter(entry -> entry.getKey() >= startTime)
                .mapToInt(Map.Entry::getValue)
                .sum();
    }
}

滑动日志能够避免突发流量，实现较为精准的限流；同样更加灵活，能够支持更加复杂的限流策略，如多级限流，每分钟不超过100次，每小时不超过300次，每天不超过1000次，我们只需要保存最近24小时所有的请求日志即可实现。

灵活并不是没有代价的，带来的缺点就是占用存储空间要高于其他限流算法。

06分布式限流

以上几种限流算法的实现都仅适合单机限流。虽然给每台机器平均分配限流配额可以达到限流的目的，但是由于机器性能，流量分布不均以及计算数量动态变化等问题，单机限流在分布式场景中的效果总是差强人意。

分布式限流最简单的实现就是利用中心化存储，即将单机限流存储在本地的数据存储到同一个存储空间中，如常见的Redis等。

当然也可以从上层流量入口进行限流，Nginx代理服务就提供了限流模块，同样能够实现高性能，精准的限流，其底层是漏桶算法。

以上がJava における一般的な電流制限アルゴリズムは何ですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。