Java分布式限流：基于Redis的滚动窗口与退避机制实现指南

<dependency>
    <groupId>com.github.vladimir-bukhtoyarov</groupId>
    <artifactId>bucket4j-core</artifactId>
    <version>8.1.1</version>
</dependency>
<dependency>
    <groupId>com.github.vladimir-bukhtoyarov</groupId>
    <artifactId>bucket4j-redis</artifactId>
    <version>8.1.1</version>
</dependency>
<!-- 如果使用Jedis客户端 -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>5.1.0</version>
</dependency>
<!-- 如果使用Lettuce客户端 -->
<!--
<dependency>
    <groupId>io.lettuce</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>6.3.2.RELEASE</version>
</dependency>
-->

import io.bucket4j.Bucket;
import io.bucket4j.ConsumptionProbe;
import io.bucket4j.grid.GridBucket;
import io.bucket4j.grid.GridBucketState;
import io.bucket4j.grid.jcache.JCacheBucketBuilder;
import io.bucket4j.redis.lettuce.Bucket4jLettuce; // 或 Bucket4jJedis
import io.lettuce.core.RedisClient; // 或 JedisPool
import io.lettuce.core.api.StatefulRedisConnection; // 或 Jedis

import javax.cache.Cache;
import javax.cache.CacheManager;
import javax.cache.Caching;
import javax.cache.configuration.MutableConfiguration;
import javax.cache.expiry.CreatedExpiryPolicy;
import javax.cache.expiry.Duration;
import javax.cache.spi.CachingProvider;
import java.util.concurrent.TimeUnit;

public class RedisRateLimiter {

    private final Bucket bucket;
    private static final String BUCKET_KEY_PREFIX = "rate-limiter:";

    public RedisRateLimiter(String redisUri, String limiterKey, long capacity, long refillTokens, long refillPeriodSeconds) {
        // 使用Lettuce客户端作为示例
        RedisClient redisClient = RedisClient.create(redisUri);
        StatefulRedisConnection<byte[], byte[]> connection = redisClient.connect(new ByteArrayCodec());

        // Bucket4j的Redis集成需要一个CacheManager，这里我们使用JCache的简单实现
        CachingProvider cachingProvider = Caching.getCachingProvider();
        CacheManager cacheManager = cachingProvider.getCacheManager();

        MutableConfiguration<String, GridBucketState> configuration = new MutableConfiguration<>();
        configuration.setStoreByValue(false) // 重要：设置为false以避免序列化问题
                     .setTypes(String.class, GridBucketState.class)
                     .setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(Duration.ETERNAL)); // 桶状态通常不需要过期

        // 创建一个JCache实例，Bucket4j将使用它来管理Redis中的桶状态
        Cache<String, GridBucketState> cache = cacheManager.createCache("rate-limiter-cache", configuration);

        // 使用Bucket4jLettuce（或Bucket4jJedis）构建ProxyManager
        bucket = Bucket4jLettuce.builder()
                .with == 0 ? connection.sync() : connection.async()) // 使用同步或异步连接
                .build()
                .get ; // 获取JCacheProxyManager实例

        // 定义限流策略：一个令牌桶，容量为capacity，每refillPeriodSeconds秒补充refillTokens个令牌
        // 这可以模拟滚动窗口的行为，通过调整容量和补充速率来控制特定时间内的请求量
        Bandwidth limit = Bandwidth.simple(capacity, Duration.of(refillTokens, TimeUnit.SECONDS));

        // 根据limiterKey获取或创建限流桶
        this.bucket = proxyManager.builder().build(limiterKey, () -> {
            return Bucket.builder().addLimit(limit).build();
        });
    }

    /**
     * 尝试消耗一个令牌，并返回消耗结果及退避时间。
     *
     * @param tokensToConsume 尝试消耗的令牌数量，通常为1。
     * @return ConsumptionResult 包含是否成功消耗以及重试时间（如果失败）。
     */
    public ConsumptionResult tryConsume(long tokensToConsume) {
        ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(tokensToConsume);

        if (probe.isConsumed()) {
            // 令牌消耗成功
            return new ConsumptionResult(true, probe.getRemainingTokens(), 0);
        } else {
            // 令牌消耗失败，获取需要等待的时间
            long nanosToWait = probe.getNanosToWaitForRefill();
            long millisToWait = TimeUnit.NANOSECONDS.toMillis(nanosToWait);
            return new ConsumptionResult(false, probe.getRemainingTokens(), millisToWait);
        }
    }

    // 辅助类用于封装消耗结果
    public static class ConsumptionResult {
        private final boolean consumed;
        private final long remainingTokens;
        private final long retryAfterMillis; // 如果未消耗，则为需要等待的毫秒数

        public ConsumptionResult(boolean consumed, long remainingTokens, long retryAfterMillis) {
            this.consumed = consumed;
            this.remainingTokens = remainingTokens;
            this.retryAfterMillis = retryAfterMillis;
        }

        public boolean isConsumed() {
            return consumed;
        }

        public long getRemainingTokens() {
            return remainingTokens;
        }

        public long getRetryAfterMillis() {
            return retryAfterMillis;
        }

        @Override
        public String toString() {
            return "ConsumptionResult{" +
                   "consumed=" + consumed +
                   ", remainingTokens=" + remainingTokens +
                   ", retryAfterMillis=" + retryAfterMillis +
                   '}';
        }
    }

    // 用于Lettuce连接的字节数组编解码器
    private static class ByteArrayCodec implements io.lettuce.core.codec.RedisCodec<byte[], byte[]> {
        @Override
        public byte[] decodeKey(java.nio.ByteBuffer byteBuffer) {
            byte[] bytes = new byte[byteBuffer.remaining()];
            byteBuffer.get(bytes);
            return bytes;
        }

        @Override
        public byte[] decodeValue(java.nio.ByteBuffer byteBuffer) {
            byte[] bytes = new byte[byteBuffer.remaining()];
            byteBuffer.get(bytes);
            return bytes;
        }

        @Override
        public java.nio.ByteBuffer encodeKey(byte[] bytes) {
            return java.nio.ByteBuffer.wrap(bytes);
        }

        @Override
        public java.nio.ByteBuffer encodeValue(byte[] bytes) {
            return java.nio.ByteBuffer.wrap(bytes);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 示例用法
        // Redis URI: redis://localhost:6379
        RedisRateLimiter limiter = new RedisRateLimiter("redis://localhost:6379", "user:123:api", 10, 10, 60); // 每分钟10个请求，容量10

        System.out.println("--- 模拟正常请求 ---");
        for (int i = 0; i < 12; i++) {
            ConsumptionResult result = limiter.tryConsume(1);
            System.out.println("请求 " + (i + 1) + ": " + result);
            if (!result.isConsumed()) {
                System.out.println("被限流，请等待 " + result.getRetryAfterMillis() + " 毫秒后重试。");
                // 实际应用中，这里可以暂停或返回HTTP 429并带上Retry-After头
                Thread.sleep(result.getRetryAfterMillis() + 10); // 模拟等待
            }
        }
    }
}