在条件之后返回通量或错误

Question

我正在学习webflux的反应式编程，为此我迁移了一些代码。

例如，我试图迁移此方法：

public Set<Vaccine> getAll(Set<Long> vaccinesIds) throws EntityNotFoundException {
        if (null == vaccinesIds) {
            return null;
        }

        Set<Long> vaccinesToFind = new HashSet<>(vaccinesIds);
        vaccinesToFind.remove(null);

        Set<Vaccine> vaccines = new HashSet<>();
        vaccineRepository.findByIdIn(vaccinesToFind).forEach(vaccines::add);

        if (vaccines.size() != vaccinesToFind.size()) {
            LOG.warn("Could not find vaccines with ids: " + vaccinesToFind.removeAll(vaccines.stream().map(Vaccine::getId).collect(Collectors.toSet())));
            throw new EntityNotFoundException(VACCINE_ERROR_NOT_FOUND);
        }

        return vaccines;
    }

总结代码，如果respository返回所有请求的疫苗，则应返回结果，否则应返回错误。

为此，我这样想，但不起作用：

public Flux<Vaccine> getAll(Set<Long> vaccinesIds) {
    if (null == vaccinesIds) {
        return Flux.empty();
    }

    Set<Long> vaccinesToFind = new HashSet<>(vaccinesIds);

    Flux<Vaccine> byIdIn = vaccineRepository.findByIdIn(vaccinesToFind);
        
    Mono<Long> filter = vaccineRepository.findByIdIn(vaccinesToFind).count().filter(x -> x.equals(Long.valueOf(vaccinesToFind.size())));

   return filter.flatMapMany(asd -> vaccineRepository.findByIdIn(vaccinesToFind)
    ).switchIfEmpty(Flux.error((new EntityNotFoundException(VACCINE_ERROR_NOT_FOUND))));
   
}

我做错了什么？

我的第一个疑问是，如果过滤器最后有一个equals方法，为什么它是一个Long的Mono。我的问题是评估过滤器以返回列表或错误。

Answer 1

首先，您正在多次查询相同的结果疫苗epository.findByIdIn（疫苗查找）。相同的数据被多次查询、传输和反序列化。这是这里出问题的迹象。

让我们假设结果集适合内存。然后想法是将通量转换为通常的集合，并决定是否发出错误：

return vaccineRepository.findByIdIn(vaccinesIds)
.collectList()
.flatMapMany(result -> {
    if(result.size() == vaccinesIds.size()) return Flux.fromIterable(result);
    else return Flux.error(new EntityNotFoundException(VACCINE_ERROR_NOT_FOUND));
});

如果结果对于主内存来说太大，您可以通过第一次查询在数据库中进行计数，在肯定的情况下查询结果。解决方案与您的代码类似：

return vaccineRepository.countByIdIn(vaccinesIds)
.filter(count -> count == vaccinesIds.size())
.flatMapMany($ -> vaccineRepository.findByIdIn(vaccinesIds))
.switchIfEmpty(Mono.error(new EntityNotFoundException(VACCINE_ERROR_NOT_FOUND)));

filter的结果是Mono