基于Redis和Bucket4j的Java分布式限流器：实现滚动窗口与回退机制

<dependencies>
    <!-- Bucket4j Core -->
    <dependency>
        <groupId>com.github.vladimir-bukhtoyarov</groupId>
        <artifactId>bucket4j-core</artifactId>
        <version>8.1.1</version> <!-- 请使用最新稳定版本 -->
    </dependency>
    <!-- Bucket4j Jedis integration -->
    <dependency>
        <groupId>com.github.vladimir-bukhtoyarov</groupId>
        <artifactId>bucket4j-redis</artifactId>
        <version>8.1.1</version> <!-- 确保与bucket4j-core版本一致 -->
    </dependency>
    <!-- Jedis client -->
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>4.3.1</version> <!-- 请使用最新稳定版本 -->
    </dependency>
</dependencies>

import io.github.bucket4j.Bandwidth;
import io.github.bucket4j.Bucket;
import io.github.bucket4j.Bucket4j;
import io.github.bucket4j.ConsumptionProbe;
import io.github.bucket4j.redis.jedis.JedisBasedProxyManager;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

public class DistributedRateLimiter {

    private static JedisPool jedisPool;
    private static JedisBasedProxyManager proxyManager;

    public static void initRedis(String redisHost, int redisPort) {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(10); // 最大连接数
        poolConfig.setMaxIdle(5);   // 最大空闲连接数
        jedisPool = new JedisPool(poolConfig, redisHost, redisPort);
        proxyManager = new JedisBasedProxyManager(jedisPool);
    }

    /**
     * 获取或创建限流桶。
     * 每个唯一的key对应一个独立的限流桶。
     *
     * @param key 限流的唯一标识符（例如：用户ID, IP地址, API路径）
     * @param capacity 令牌桶容量
     * @param refillTokens 每次填充的令牌数
     * @param refillPeriod 令牌填充周期
     * @return 配置好的限流桶
     */
    public static Bucket getOrCreateLimiterBucket(String key, long capacity, long refillTokens, Duration refillPeriod) {
        // 定义限流带宽
        Bandwidth limit = Bandwidth.builder()
                .capacity(capacity) // 令牌桶容量
                .refillGreedy(refillTokens, refillPeriod) // 令牌填充策略：每 refillPeriod 填充 refillTokens 个令牌
                .build();

        // 使用proxyManager获取或创建分布式桶
        // key是限流器的唯一标识，Supplier是当key不存在时如何创建桶的定义
        return Bucket4j.builder()
                .addLimit(limit)
                .build(proxyManager, key);
    }

    public static void main(String[] args) throws InterruptedException {
        // 初始化Redis连接池
        initRedis("localhost", 6379); // 假设Redis运行在本地默认端口

        String userId = "user:123";
        // 为该用户设置限流：每秒10个请求，最大突发20个请求
        Bucket userBucket = getOrCreateLimiterBucket(userId, 20, 10, Duration.ofSeconds(1));

        System.out.println("--- 模拟用户请求 ---");
        for (int i = 0; i < 30; i++) {
            // 尝试消费一个令牌
            ConsumptionProbe probe = userBucket.tryConsumeAndReturnRemaining(1);

            if (probe.isConsumed()) {
                // 请求成功
                System.out.printf("请求 %d 成功！剩余令牌：%d%n", i + 1, probe.getRemainingTokens());
            } else {
                // 请求被拒绝
                long nanosToWaitForRefill = probe.getNanosToWaitForRefill();
                long millisToWaitForRefill = TimeUnit.NANOSECONDS.toMillis(nanosToWaitForRefill);
                System.err.printf("请求 %d 被拒绝！需要等待 %d 毫秒后重试。当前剩余令牌：%d%n",
                        i + 1, millisToWaitForRefill, probe.getRemainingTokens());

                // 模拟客户端等待回退时间
                if (millisToWaitForRefill > 0) {
                    Thread.sleep(millisToWaitForRefill);
                }
            }
            // 每次请求间隔模拟
            Thread.sleep(50);
        }

        // 关闭Redis连接池
        jedisPool.close();
    }
}