Redis不可用时Bucket4j tryConsume方法阻塞问题及方案咨询

问题根源 #

1. Lettuce同步操作阻塞：Redis离线时，redisBucket.tryConsume()调用的是Lettuce同步API，默认超时时间较长（通常几十秒），会直接阻塞请求线程，直到超时才抛出异常，此时定时监控的切换逻辑根本赶不上。
2. 线程安全隐患：currentBucket和redisAvailable未做线程同步，定时任务和请求线程同时操作时可能出现竞态条件。
3. 被动切换不及时：依赖3秒一次的定时检测切换Bucket，无法处理检测间隔内的Redis故障。

修正方案 #

1. 配置Lettuce超时，缩短阻塞时间 #

给Lettuce客户端设置命令超时和连接超时，避免请求长时间阻塞。

2. 线程安全的Bucket切换 #

用AtomicReference存储当前Bucket，保证多线程环境下的原子性操作；redisAvailable用volatile修饰，保证可见性。

3. 主动捕获异常，即时切换 #

在consume()方法中捕获Redis相关异常，一旦触发立即切换到本地Bucket，同时标记Redis不可用。

4. 优化监控恢复逻辑 #

定时任务检测Redis恢复时，将Bucket切回Redis版本，同时重置状态。

修正后的完整代码 #

import io.github.bucket4j.Bandwidth;
import io.github.bucket4j.Bucket;
import io.github.bucket4j.Bucket4j;
import io.github.bucket4j.BucketConfiguration;
import io.github.bucket4j.Refill;
import io.github.bucket4j.distributed.proxy.ExpirationAfterWriteStrategy;
import io.github.bucket4j.redis.lettuce.LettuceBasedProxyManager;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisConnectionException;
import io.lettuce.core.RedisURI;
import io.lettuce.core.TimeoutOptions;
import io.lettuce.core.api.StatefulRedisConnection;
import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class ResilientRateLimiter {

    private final Bucket redisBucket;
    private final Bucket localBucket;
    private final AtomicReference<Bucket> currentBucket;
    private volatile boolean redisAvailable = true;

    private static final String REDIS_URI = "redis://localhost:6379";
    // 缩短超时时间，避免长时间阻塞
    private static final Duration REDIS_COMMAND_TIMEOUT = Duration.ofMillis(500);

    public ResilientRateLimiter() {
        RedisClient redisClient = buildRedisClient();
        LettuceBasedProxyManager proxyManager = buildProxyManager(redisClient);

        // Redis bucket with max 50 TPS
        Refill refill = Refill.greedy(50, Duration.ofSeconds(1));
        Bandwidth limit = Bandwidth.classic(50, refill);
        BucketConfiguration configuration = BucketConfiguration.builder().addLimit(limit).build();
        this.redisBucket = proxyManager.builder().build("123".getBytes(), configuration);

        // Local bucket with max 5 TPS
        Bandwidth localLimit = Bandwidth.simple(5, Duration.ofSeconds(1));
        this.localBucket = Bucket4j.builder().addLimit(localLimit).build();

        this.currentBucket = new AtomicReference<>(redisBucket);
        
        startMonitoring(redisClient);
    }

    private RedisClient buildRedisClient() {
        RedisURI redisURI = RedisURI.create(REDIS_URI);
        // 设置连接和命令超时
        redisURI.setTimeout(REDIS_COMMAND_TIMEOUT);
        return RedisClient.create(redisURI);
    }

    private static LettuceBasedProxyManager buildProxyManager(RedisClient redisClient) {
        return LettuceBasedProxyManager.builderFor(redisClient)
                .withExpirationStrategy(ExpirationAfterWriteStrategy.
                        basedOnTimeForRefillingBucketUpToMax(Duration.ofSeconds(2)))
                // 给代理管理器也设置超时选项
                .withTimeoutOptions(TimeoutOptions.builder()
                        .fixedTimeout(REDIS_COMMAND_TIMEOUT)
                        .build())
                .build();
    }

    private void switchToLocalBucket() {
        if (currentBucket.compareAndSet(redisBucket, localBucket)) {
            System.out.println("Switching to local bucket with 5 TPS");
            redisAvailable = false;
        }
    }

    private void switchToRedisBucket() {
        if (currentBucket.compareAndSet(localBucket, redisBucket)) {
            System.out.println("Switching back to Redis bucket with 50 TPS");
            redisAvailable = true;
        }
    }

    private void startMonitoring(RedisClient redisClient) {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(() -> {
            try (StatefulRedisConnection<String, String> connection = redisClient.connect()) {
                connection.sync().ping(); // Test redis connection
                if (!redisAvailable) {
                    switchToRedisBucket();
                }
            } catch (RedisConnectionException e) {
                System.err.println("Redis is unavailable: " + e.getMessage());
                if (redisAvailable) {
                    switchToLocalBucket();
                }
            }
        }, 0, 3, TimeUnit.SECONDS); // wait 3 seconds until next verification
    }

    public boolean consume() {
        try {
            return currentBucket.get().tryConsume(1);
        } catch (RuntimeException e) {
            // 捕获Redis相关异常（Bucket4j封装的Redis异常通常是RuntimeException）
            if (e.getCause() instanceof RedisConnectionException || 
                e.getMessage().contains("Redis") || e.getMessage().contains("connection")) {
                switchToLocalBucket();
                // 切换后用本地Bucket处理当前请求
                return localBucket.tryConsume(1);
            }
            // 非Redis异常，直接抛出
            throw e;
        }
    }
}

关键改动说明 #

• Lettuce超时配置：在构建RedisClient和ProxyManager时都设置了500ms的超时，避免请求长时间阻塞。
• AtomicReference管理Bucket：用compareAndSet保证切换操作的原子性，避免多线程下的竞态问题。
• consume方法异常捕获：主动捕获Redis相关异常，触发即时切换，同时用本地Bucket处理当前请求，避免请求失败。
• volatile修饰redisAvailable：保证多线程下状态的可见性，避免定时任务和请求线程的状态不一致。

更优实现思路 #

如果需要更优雅的故障切换，可以考虑：

• 使用Lettuce异步API：将Redis Bucket的操作改为异步，配合超时回调，完全避免阻塞。
• 熔断机制：结合Resilience4j实现熔断，当Redis故障次数达到阈值后自动切换到本地Bucket，恢复时自动切回。
• 本地Bucket的动态调整：根据Pod数量动态调整本地Bucket的限流阈值，避免多Pod下总流量超过预期。

内容的提问来源于stack exchange，提问作者elias