Spring Boot集成Redis缓存Flux对象时序列化失败问题求助

你遇到的序列化错误核心原因是控制器上的@Cacheable注解尝试缓存Flux<WalletCharges>对象。Spring Cache的默认机制是为同步场景设计的，不支持直接序列化Reactor的响应式发布者（Flux/Mono），它期望缓存的是可序列化的业务实体，而非响应式流本身。你已经通过RedisWalletChargesRepository实现了响应式缓存逻辑，控制器上的@Cacheable属于重复操作，反而触发了序列化异常。

步骤1：移除控制器上的@Cacheable注解 #

直接删除WalletChargesController类上的@Cacheable，保留你手动实现的响应式缓存逻辑即可。

步骤2：修复服务类的缓存写入逻辑 #

当前getWalletChargesFromDBAndCache方法中，调用redisWalletChargesRepository.save()后未处理返回的Flux，可能导致缓存写入未完成就返回数据。需要将缓存操作与数据返回链式绑定，确保缓存写入完成后再响应结果。

步骤3：优化Redis仓库的异常处理 #

避免直接返回null，改用Flux.error()传递异常，符合响应式编程规范。

修改后的代码 #

WalletChargesController.java #

@CrossOrigin(origins = {"https://localhost:3000", "http://localhost:3000"} )
@RestController
@RequestMapping("v1/walletcharges")
public class WalletChargesController {

    private final WalletChargesService walletChargesService;

    @Autowired
    public WalletChargesController(WalletChargesService walletChargesService) {
        this.walletChargesService = walletChargesService;
    }
    
    @GetMapping
    public Flux<WalletCharges> getWalletCharges() {
        return walletChargesService.getWalletCharges();
    }
}

WalletChargesServiceImpl.java #

@Service
public class WalletChargesServiceImpl implements WalletChargesService {

    private final WalletChargesRepository repository;
    private final RedisWalletChargesRepository redisWalletChargesRepository;

    @Autowired
    public WalletChargesServiceImpl(WalletChargesRepository repository, RedisWalletChargesRepository redisWalletChargesRepository) {
        this.repository = repository;
        this.redisWalletChargesRepository = redisWalletChargesRepository;
    }

    @Override
    public Flux<WalletCharges> getWalletCharges() {
        return redisWalletChargesRepository
                .findByKey("allWalletCharges")
                .switchIfEmpty(Flux.defer(this::getWalletChargesFromDBAndCache));
    }

    @Override
    public Flux<WalletCharges> getWalletChargesFromDBAndCache(){
        System.out.println("The getWalletChargesFromDBAndCache method was invoked");
        List<WalletCharges> walletCharges = repository.findAll();
        // 链式绑定缓存写入与数据返回，确保缓存操作完成
        return redisWalletChargesRepository.save("allWalletCharges", walletCharges)
                .switchIfEmpty(Flux.fromIterable(walletCharges));
    }
}

RedisWalletChargesRepository.java #

@Repository
public class RedisWalletChargesRepository {
    
    private final ReactiveRedisOperations<String, String> operations;

    public RedisWalletChargesRepository(ReactiveRedisOperations<String, String> operations) {
        this.operations = operations;
    }

    public Flux<WalletCharges> save(String key, List<WalletCharges> walletCharges) {
        try {
            String walletChargesString = ObjectMapperUtils.convertToString(walletCharges);
            return operations.opsForValue()
                    .set(key, walletChargesString)
                    .flatMapMany(__ -> Flux.fromIterable(walletCharges));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            return Flux.error(e); // 用响应式方式传递异常
        }
    }

    public Flux<WalletCharges> findByKey(String key) {
        return operations.opsForValue()
                .get(key)
                .flatMapMany(result ->  {
                    try {
                        List<WalletCharges> walletCharges = ObjectMapperUtils.convertToObject(result);
                        return Flux.fromIterable(walletCharges);
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                        return Flux.error(e); // 用响应式方式传递异常
                    }
                })
                .switchIfEmpty(Flux.empty()); // 未找到数据时返回空流
    }
}

额外说明 #

1. 响应式编程中不要忽略返回的Flux/Mono，必须通过链式调用保证操作的顺序和完成性。
2. 你的Flux.fromIterable(walletCharges)用法是正确的，这不是导致序列化错误的原因，无需修改。
3. 实体类WalletCharges实现Serializable是正确的，配合ObjectMapper序列化为字符串存储Redis的逻辑没问题。

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