Spring Data Redis Reactive：ReactiveRedisTemplate的Pipeline批量操作方案咨询

在响应式Redis客户端模型中，没有和RedisTemplate.executePipelined()完全对应的直接API，但可以通过以下几种方式实现等价的批量/低网络开销操作：

1. 使用ReactiveRedisConnection的批量命令打包 #

这是最接近Pipeline的实现，底层依赖Lettuce的Batch API，会将多个命令打包成一组请求发送，减少网络往返次数。

@Autowired
private ReactiveRedisTemplate<String, String> reactiveRedisTemplate;

public Flux<String> executeBatch() {
    return reactiveRedisTemplate.execute(connection -> {
        ReactiveStringCommands stringCommands = connection.stringCommands();
        
        return connection.createBatch()
                // 批量加入需要执行的命令
                .queue(stringCommands.set(ByteBuffer.wrap("key1".getBytes()), ByteBuffer.wrap("val1".getBytes())))
                .queue(stringCommands.get(ByteBuffer.wrap("key1".getBytes())))
                .queue(stringCommands.set(ByteBuffer.wrap("key2".getBytes()), ByteBuffer.wrap("val2".getBytes())))
                .queue(stringCommands.get(ByteBuffer.wrap("key2".getBytes())))
                // 执行批量操作并返回结果流
                .execute()
                // 转换结果格式
                .map(buf -> new String(buf.array()));
    });
}

2. 利用响应式流组合操作 #

如果不需要严格的命令打包，可通过Flux.merge()或Flux.concat()组合多个命令的响应式实例，实现并行/串行的批量处理：

public Flux<String> combineReactiveOperations() {
    Mono<String> setKey1 = reactiveRedisTemplate.opsForValue().set("key1", "val1").thenReturn("key1 已设置");
    Mono<String> getKey1 = reactiveRedisTemplate.opsForValue().get("key1");
    Mono<String> setKey2 = reactiveRedisTemplate.opsForValue().set("key2", "val2").thenReturn("key2 已设置");
    Mono<String> getKey2 = reactiveRedisTemplate.opsForValue().get("key2");

    // 并行执行所有操作，结果顺序不确定
    return Flux.merge(setKey1, getKey1, setKey2, getKey2);
    // 若需严格保持命令执行顺序，改用concat
    // return Flux.concat(setKey1, getKey1, setKey2, getKey2);
}

3. 使用Redis原生批量命令 #

优先使用Redis自带的批量命令（如MSET/MGET/HMSET等），这类命令本身就是为批量场景设计的，性能比Pipeline更优：

// 批量设置键值对
public Mono<Boolean> batchSetValues() {
    Map<String, String> keyValueMap = new HashMap<>();
    keyValueMap.put("key1", "val1");
    keyValueMap.put("key2", "val2");
    keyValueMap.put("key3", "val3");
    
    return reactiveRedisTemplate.opsForValue().multiSet(keyValueMap);
}

// 批量获取键值
public Flux<String> batchGetValues() {
    List<String> keys = Arrays.asList("key1", "key2", "key3");
    return reactiveRedisTemplate.opsForValue().multiGet(keys).flatMapMany(Flux::fromIterable);
}

关键提示 #

• ReactiveRedisConnection.createBatch()是响应式模型下等价于Pipeline的核心API，适用于混合多种命令的批量场景；
• 优先选用Redis原生批量命令，避免不必要的Pipeline封装；
• 所有方案均适配spring-data-redis 2.7.5 + Lettuce的版本组合。

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