如何在单事务中实现Reactive Redis与Postgres的联动保存？

你当前代码里的@Transactional只能管控PostgreSQL的事务，Redis作为非关系型数据库，不参与Spring的声明式事务体系，所以没法自动实现两者的原子性。要达成任一操作失败都回滚的目标，得通过手动协调事务流程+补偿机制来实现：

1. 放弃声明式@Transactional，改用编程式事务管理，手动控制Postgres事务的提交/回滚
2. 串联两个操作的执行逻辑，任一环节失败时触发对应回滚操作
3. 针对极端异常场景（比如Redis成功但Postgres提交失败），添加Redis数据补偿逻辑

修改后的代码示例：

private final ReactiveRedisOperations<String, RedisLocation> locationOps;
private final ReactiveTransactionManager transactionManager; // 注入Postgres对应的反应式事务管理器
private final LocationRepository locationRepository;

// 移除原方法上的@Transactional，改用编程式事务控制
public Mono<Location> saveLocationToDbAndRedis(Location location) {
    // 创建事务操作器
    TransactionalOperator transactionalOperator = TransactionalOperator.create(transactionManager);
    
    return transactionalOperator.execute(status -> 
        // 第一步：执行Postgres保存（此时事务未提交）
        locationRepository.save(location)
            .flatMap(savedLocation -> 
                // 第二步：执行Redis保存操作
                saveAndUpdateLocationInRedis(savedLocation)
                    .flatMap(redisSuccess -> {
                        if (!redisSuccess) {
                            // Redis保存失败，标记Postgres事务回滚
                            status.setRollbackOnly();
                            return Mono.error(new RuntimeException("Redis保存失败，触发Postgres回滚"));
                        }
                        // Redis保存成功，返回结果
                        return Mono.just(savedLocation);
                    })
                    // 捕获Redis操作异常，触发Postgres回滚
                    .onErrorResume(e -> {
                        status.setRollbackOnly();
                        return Mono.error(new RuntimeException("Redis操作异常，触发Postgres回滚", e));
                    })
            )
            // 捕获Postgres保存异常，直接抛出，Redis不会执行
            .onErrorResume(e -> Mono.error(new RuntimeException("Postgres保存失败", e)))
    )
    // 处理事务提交失败的极端情况：此时Redis已保存数据，需要删除做补偿
    .onErrorResume(e -> 
        saveAndDeleteLocationInRedis(location.getId())
            .then(Mono.error(new RuntimeException("事务提交失败，已补偿Redis数据", e)))
    );
}

// 原Redis保存方法保持不变
public Mono<Boolean> saveAndUpdateLocationInRedis(Location location) {
    return locationOps.opsForHash().put(
            REDIS_HASH_LOCATION_KEY,
            location.getId(),
            toRedisLocation(location)
    );
}

// 新增Redis删除方法，用于异常场景下的补偿
public Mono<Boolean> saveAndDeleteLocationInRedis(Long locationId) {
    return locationOps.opsForHash().delete(REDIS_HASH_LOCATION_KEY, locationId);
}

关键注意点 #

• 确保注入的transactionManager是PostgreSQL对应的反应式事务管理器（比如R2dbcTransactionManager）
• 补偿逻辑是为了覆盖“Redis成功但Postgres提交失败”的极端场景，保证数据最终一致性
• 所有异常分支都要明确触发回滚或补偿，避免出现数据不一致的情况

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