如何在Quarkus中使用Redis Pipeline解决连接池队列溢出问题？

问题分析 #

你当前循环调用hget的方式，每个请求都会单独和Redis建立交互，在高并发场景下会快速耗尽连接池资源，导致等待队列溢出。通过Redis Pipeline批量打包多个命令，可以大幅减少TCP往返次数，降低连接池压力。

实现步骤 #

1. 修改CacheService，添加批量获取方法

利用Quarkus Redis Client提供的batch()接口，创建哈希命令的批量实例，一次性打包所有hget请求：

import java.util.List;

public class CacheService {

    private final HashCommands<String, String, CachedData> redisHash;

    protected CacheService(RedisDataSource redis) {
        this.redisHash = redis.hash(CachedData.class);
    }

    protected CachedData get(String name) {
        return redisHash.hget("data", name);
    }

    // 新增批量获取方法
    protected List<CachedData> batchGet(List<String> names) {
        // 创建哈希批量命令实例
        HashBatchCommands<String, String, CachedData> batch = redisHash.batch();
        // 将所有hget命令加入批量队列
        names.forEach(name -> batch.hget("data", name));
        // 执行批量命令并返回所有结果
        return batch.execute().results();
    }
}

2. 替换原有循环逻辑

直接调用新增的batchGet方法，替代原来的流式循环：

List<CachedData> dataList = cacheService.batchGet(names);

进阶优化：大列表分批次处理 #

如果names列表过大（比如超过1000条），单个Pipeline包体积会过大，可能引发Redis性能问题。可以添加分批次处理逻辑：

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class CacheService {
    // ... 原有代码 ...

    protected List<CachedData> batchGet(List<String> names, int batchSize) {
        return IntStream.range(0, (names.size() + batchSize - 1) / batchSize)
                .mapToObj(i -> names.subList(
                        i * batchSize,
                        Math.min((i + 1) * batchSize, names.size())
                ))
                .flatMap(subList -> {
                    HashBatchCommands<String, String, CachedData> batch = redisHash.batch();
                    subList.forEach(name -> batch.hget("data", name));
                    return batch.execute().results().stream();
                })
                .collect(Collectors.toList());
    }
}

调用时指定批次大小（比如100）：

List<CachedData> dataList = cacheService.batchGet(names, 100);

原理说明 #

Quarkus Redis Client的batch机制会自动将多个命令打包成Pipeline请求，通过单个连接一次性发送给Redis，Redis处理后一次性返回所有结果。这种方式能将N次TCP往返减少为1次（或分批次的k次），极大降低连接池的资源消耗，避免高并发下的等待队列溢出问题。

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