Quarkus Redis缓存：如何获取连接池使用量实现背压机制

1. 获取Redis连接池状态 #

Quarkus的Redis缓存基于Vert.x Redis客户端实现，你可以通过连接池的PoolMetrics获取等待队列长度、活跃连接数等关键指标：

注入连接池并获取指标 #

注入RedisConnectionPool（Quarkus 2.x+版本均支持），调用其metrics()方法即可拿到池状态数据：

@Inject
RedisConnectionPool redisConnectionPool;

@Inject
@CacheName("myCache")
Cache myCache;

public void insertIntoMyCache(UUID messageKey, String value) {
    // 提前检查连接池状态，触发前置背压
    PoolMetrics poolMetrics = redisConnectionPool.metrics();
    int waitingRequests = poolMetrics.waiting();
    int activeConnections = poolMetrics.active();
    int maxPoolSize = poolMetrics.maxSize();

    // 自定义触发阈值：比如等待请求超160，或活跃连接占比超80%
    boolean needBackpressure = waitingRequests > 160 || (activeConnections * 100 / maxPoolSize) > 80;
    if (needBackpressure) {
        pauseKafkaConsumer();
    }

    // 执行缓存写入
    myCache.as(RedisCache.class).put(messageKey, value).subscribe().with(
            success -> log.info("Successfully inserted key {}.", messageKey),
            error -> {
                log.error("Error while inserting message into cache. Key {}", messageKey);
                // 捕获连接池异常，补充触发背压
                if (error instanceof ConnectionPoolTooBusyException) {
                    pauseKafkaConsumer();
                }
            }
    );
}

PoolMetrics核心方法说明：

• waiting(): 当前等待连接的请求数
• active(): 当前活跃的连接数
• maxSize(): 连接池最大容量（对应配置quarkus.redis.max-pool-size）
• idle(): 空闲连接数

2. 实现Kafka消费者暂停逻辑 #

根据你使用的Kafka消费方式，选择对应的暂停方案：

方式1：Reactive Messaging场景（@Incoming注解） #

通过延迟处理消息实现软背压：

@Incoming("kafka-topic")
public CompletionStage<Void> consumeMessage(Message<String> message) {
    if (needBackpressure()) {
        // 延迟500ms再处理，给连接池腾空间
        return CompletableFuture.delayedExecutor(500, TimeUnit.MILLISECONDS)
                .submit(() -> processMessage(message))
                .thenCompose(v -> message.ack());
    }
    return processMessage(message).thenCompose(v -> message.ack());
}

方式2：传统Kafka Consumer API场景 #

直接操作消费者实例暂停/恢复：

@Inject
KafkaConsumer<String, String> kafkaConsumer;

private void pauseKafkaConsumer() {
    Set<TopicPartition> partitions = kafkaConsumer.assignment();
    kafkaConsumer.pause(partitions);
    // 1秒后自动恢复消费
    CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)
            .submit(() -> kafkaConsumer.resume(partitions));
}

额外注意事项 #

• 你的配置中quarkus.redis-max-pool-waiting存在拼写错误，正确配置项应为quarkus.redis.max-pool-waiting
• 建议将背压阈值与连接池配置联动（比如等待阈值设为max-pool-waiting的80%），避免硬编码
• 异步写入缓存时，背压逻辑建议放在统一调度器中执行，保证线程安全

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