Java中的并发限流算法：令牌桶、漏桶的精准实现与优化 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Java 并发限流算法：令牌桶、漏桶的精准实现与优化

大家好，今天我们来深入探讨 Java 中并发限流的两种经典算法：令牌桶和漏桶。在构建高并发系统时，限流是保证系统稳定性和可用性的关键手段。它可以防止突发流量压垮系统，保证服务质量。我们将从理论概念出发，逐步实现这两种算法，并探讨其优化策略。

1. 限流的必要性与常见策略

在讨论具体算法之前，我们先明确为什么需要限流。在高并发场景下，如果请求量超过系统处理能力，会导致服务响应变慢、甚至崩溃。限流就是为了避免这种情况发生，它通过限制单位时间内请求的速率，保证系统在高负载下依然能够正常运行。

常见的限流策略包括：

计数器法： 简单粗暴，在单位时间内记录请求次数，超过阈值则拒绝请求。缺点是无法应对时间窗口边界的突发流量。
滑动窗口： 改进的计数器法，将时间窗口划分为多个更小的时间段，分别计数，统计时滑动窗口，更加平滑，但实现相对复杂。
令牌桶： 以恒定速率生成令牌，请求只有拿到令牌才能通过。可以应对突发流量，允许一定程度的 burst。
漏桶： 请求进入漏桶，以恒定速率流出。平滑流量，防止突发流量压垮系统。

今天我们重点讲解令牌桶和漏桶算法。

2. 令牌桶算法

2.1 算法原理

令牌桶算法的核心思想是以恒定的速率向桶中放入令牌，每个请求需要从桶中获取一个令牌才能通过。如果桶中没有令牌，请求将被拒绝或等待。

令牌生成速率 (rate)： 每秒/毫秒/分钟生成的令牌数量。
桶的容量 (capacity)： 桶中最多能存放的令牌数量。
请求处理： 每个请求尝试从桶中获取令牌，获取成功则放行，否则拒绝或等待。

2.2 代码实现 (Java)

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class TokenBucket {

    private final int capacity;
    private final double rate; // 每秒生成的令牌数
    private AtomicInteger tokens;
    private long lastRefillTimestamp;

    public TokenBucket(int capacity, double rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.tokens = new AtomicInteger(capacity);
        this.lastRefillTimestamp = System.nanoTime();
    }

    public synchronized boolean tryAcquire(int requestedTokens) {
        refill(); // 尝试补充令牌
        if (tokens.get() >= requestedTokens) {
            tokens.getAndAdd(-requestedTokens);
            return true;
        }
        return false;
    }

    private void refill() {
        long now = System.nanoTime();
        double elapsedTime = (double) (now - lastRefillTimestamp) / 1_000_000_000.0; // 纳秒转换为秒
        int newTokens = (int) (elapsedTime * rate);

        if (newTokens > 0) {
            tokens.getAndAccumulate(newTokens, (prev, x) -> Math.min(capacity, prev + x));
            lastRefillTimestamp = now;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        TokenBucket tokenBucket = new TokenBucket(10, 2); // 容量为10，每秒生成2个令牌

        // 模拟并发请求
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                if (tokenBucket.tryAcquire(1)) {
                    System.out.println(Thread.currentThread().getName() + ": Request processed");
                } else {
                    System.out.println(Thread.currentThread().getName() + ": Request rejected");
                }
            }).start();
            Thread.sleep(100); // 模拟请求间隔
        }
    }
}

2.3 代码解释

capacity: 令牌桶的容量。
rate: 令牌生成速率，单位为每秒生成的令牌数。
tokens: 当前令牌桶中的令牌数量，使用 AtomicInteger 保证线程安全。
lastRefillTimestamp: 上次补充令牌的时间戳。
tryAcquire(int requestedTokens): 尝试获取指定数量的令牌。首先调用 refill() 方法补充令牌，然后判断令牌桶中是否有足够的令牌。如果足够，则获取令牌并返回 true，否则返回 false。
refill(): 补充令牌的方法。根据上次补充令牌的时间戳和当前时间戳计算出需要补充的令牌数量，并将令牌添加到令牌桶中。如果添加后的令牌数量超过容量，则只添加至容量上限。
synchronized: 保证 tryAcquire 方法的线程安全.

2.4 优化策略

定时任务补充令牌： 可以使用 ScheduledExecutorService 定时补充令牌，避免每次请求都计算需要补充的令牌数量。

public class TokenBucketScheduled {

    private final int capacity;
    private final double rate; // 每秒生成的令牌数
    private AtomicInteger tokens;

    public TokenBucketScheduled(int capacity, double rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.tokens = new AtomicInteger(capacity);

        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(this::refill, 0, (long) (1000/rate), TimeUnit.MILLISECONDS); // 每隔 1/rate 秒补充一次令牌
    }

    public boolean tryAcquire(int requestedTokens) {
        if (tokens.get() >= requestedTokens) {
            return tokens.compareAndSet(tokens.get(), tokens.get() - requestedTokens);
        }
        return false;
    }

    private void refill() {
        tokens.getAndAccumulate(1, (prev, x) -> Math.min(capacity, prev + x));
    }

    public static void main(String[] args) throws InterruptedException {
        TokenBucketScheduled tokenBucket = new TokenBucketScheduled(10, 2); // 容量为10，每秒生成2个令牌

        // 模拟并发请求
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                if (tokenBucket.tryAcquire(1)) {
                    System.out.println(Thread.currentThread().getName() + ": Request processed");
                } else {
                    System.out.println(Thread.currentThread().getName() + ": Request rejected");
                }
            }).start();
            Thread.sleep(100); // 模拟请求间隔
        }

        Thread.sleep(5000); // 等待一段时间，让令牌桶积累一些令牌
    }
}

这个优化方案使用 ScheduledExecutorService 定时执行 refill 方法，每次补充一个令牌。这种方式避免了每次请求都计算需要补充的令牌数量，提高了效率。使用 compareAndSet 保证 tryAcquire 的原子性.

Guava RateLimiter： Google Guava 库提供了 RateLimiter 类，它实现了令牌桶算法，并且提供了更加丰富的功能，例如平滑预热等。

import com.google.common.util.concurrent.RateLimiter;

public class TokenBucketGuava {

    private final RateLimiter rateLimiter;

    public TokenBucketGuava(double permitsPerSecond) {
        this.rateLimiter = RateLimiter.create(permitsPerSecond);
    }

    public boolean tryAcquire() {
        return rateLimiter.tryAcquire();
    }

    public static void main(String[] args) throws InterruptedException {
        TokenBucketGuava tokenBucket = new TokenBucketGuava(2); // 每秒允许 2 个请求

        // 模拟并发请求
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                if (tokenBucket.tryAcquire()) {
                    System.out.println(Thread.currentThread().getName() + ": Request processed");
                } else {
                    System.out.println(Thread.currentThread().getName() + ": Request rejected");
                }
            }).start();
            Thread.sleep(100); // 模拟请求间隔
        }
    }
}

Guava 的 RateLimiter 提供了更多的灵活性和更完善的功能，例如可以设置预热期，让系统逐渐适应高负载。

3. 漏桶算法

3.1 算法原理

漏桶算法的核心思想是将所有请求放入一个固定容量的桶中，然后以恒定的速率从桶中流出请求。如果桶满了，则拒绝新的请求。

桶的容量 (capacity)： 桶中最多能存放的请求数量。
流出速率 (rate)： 每秒/毫秒/分钟流出的请求数量。
请求处理： 请求进入桶中，如果桶未满则放入，否则拒绝。桶中的请求以恒定速率流出。

3.2 代码实现 (Java)

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class LeakyBucket {

    private final int capacity;
    private final double rate; // 每秒流出的请求数量
    private final BlockingQueue<Runnable> queue;

    public LeakyBucket(int capacity, double rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.queue = new LinkedBlockingQueue<>(capacity);

        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(this::leak, 0, (long) (1000/rate), TimeUnit.MILLISECONDS); // 每隔 1/rate 秒漏出一个请求
    }

    public boolean tryAcquire(Runnable task) {
        return queue.offer(task); // 尝试将任务放入队列
    }

    private void leak() {
        Runnable task = queue.poll(); // 从队列中取出一个任务
        if (task != null) {
            task.run(); // 执行任务
        }
    }

    public static void main(String[] args) throws InterruptedException {
        LeakyBucket leakyBucket = new LeakyBucket(5, 2); // 容量为5，每秒流出2个请求

        // 模拟并发请求
        for (int i = 0; i < 20; i++) {
            int taskId = i;
            if (leakyBucket.tryAcquire(() -> System.out.println(Thread.currentThread().getName() + ": Task " + taskId + " processed"))) {
                System.out.println(Thread.currentThread().getName() + ": Task " + taskId + " accepted");
            } else {
                System.out.println(Thread.currentThread().getName() + ": Task " + taskId + " rejected");
            }
            Thread.sleep(50); // 模拟请求间隔
        }

        Thread.sleep(3000); // 等待一段时间，让桶中的请求处理完
    }
}

3.3 代码解释

capacity: 漏桶的容量。
rate: 漏桶的流出速率，单位为每秒流出的请求数。
queue: 使用 LinkedBlockingQueue 作为漏桶，存储待处理的请求。
tryAcquire(Runnable task): 尝试将任务放入队列，如果队列已满则返回 false，表示请求被拒绝。
leak(): 以恒定速率从队列中取出任务并执行。

3.4 优化策略

使用更高效的队列： LinkedBlockingQueue 适用于高并发场景，但如果对性能有更高的要求，可以考虑使用 Disruptor 等更高效的队列。
异步处理： 可以将任务放入线程池中异步执行，避免阻塞 leak() 方法。

4. 令牌桶 vs 漏桶

特性	令牌桶	漏桶
流量整形	允许突发流量，但平均速率受到限制	平滑流量，将所有流量整形为恒定速率
应用场景	应对突发流量，允许一定程度的 burst	防止突发流量压垮系统，保证服务质量
实现复杂度	相对简单	相对简单
参数	桶的容量，令牌生成速率	桶的容量，流出速率
适用性	电商秒杀、API 接口限流等，允许一定程度的 burst	消息队列、流媒体服务器等，需要平滑流量的场景

5. 选择合适的限流算法

选择哪种限流算法取决于具体的应用场景和需求。

如果需要应对突发流量，并且允许一定程度的 burst，那么令牌桶算法是一个不错的选择。
如果需要平滑流量，防止突发流量压垮系统，那么漏桶算法更适合。
在实际应用中，也可以将两种算法结合使用，例如使用令牌桶允许一定程度的 burst，然后使用漏桶平滑流量。

6. 分布式限流

以上讨论的都是单机限流，在分布式系统中，需要使用分布式锁或 Redis 等分布式缓存来实现全局限流。

Redis + Lua： 可以使用 Redis 作为计数器，并使用 Lua 脚本保证原子性。
Redisson： Redisson 是一个 Java Redis 客户端，它提供了分布式限流器 RRateLimiter。

示例 (Redisson 分布式限流)：

import org.redisson.Redisson;
import org.redisson.api.RRateLimiter;
import org.redisson.config.Config;
import java.util.concurrent.TimeUnit;

public class RedissonRateLimiterExample {

    public static void main(String[] args) throws InterruptedException {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");

        Redisson redisson = (Redisson) Redisson.create(config);

        RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter");
        rateLimiter.trySetRate(RateIntervalUnit.PER_SECOND, 2, RateType.OVERALL); // 每秒最多允许 2 个请求

        for (int i = 0; i < 10; i++) {
            if (rateLimiter.tryAcquire(1)) {
                System.out.println(Thread.currentThread().getName() + ": Request " + i + " processed");
            } else {
                System.out.println(Thread.currentThread().getName() + ": Request " + i + " rejected");
            }
            Thread.sleep(200);
        }

        redisson.shutdown();
    }
}

关于限流算法和分布式限流的总结

我们深入探讨了令牌桶和漏桶这两种经典的并发限流算法，并提供了 Java 代码实现和优化策略。同时，我们也简要介绍了分布式限流的实现方式。选择合适的限流算法需要根据具体的应用场景和需求进行权衡。希望今天的分享能够帮助大家更好地理解和应用并发限流技术，构建更加稳定和可靠的高并发系统。

Java 并发限流算法：令牌桶、漏桶的精准实现与优化

发表回复 取消回复

发表回复取消回复