如何在响应式Spring Data中应用分页？

Question

在Spring Data中，我们有继承自CrudRepository的PagingAndSortingRepository。在响应式Spring Data中，我们只有继承自ReactiveCrudRepository的ReactiveSortingRepository。 如何以响应式方式实现分页？ 未来是否能够使用例如ReactivePagingAndSortingRepository 实现分页呢？

- Steph

---

你想解决什么使用案例？是批量数据处理（逐页）还是只想获取特定的数据块？ - mp911de

第二点。我只想通过分页数据来创建一个Web服务，而不是一次性检索所有数据。 - Steph

Answer 1

响应式Spring Data MongoDB存储库在分页方面不提供命令式存储库设计的分页功能。命令式分页需要在获取页面时提供额外的细节。具体而言，需要：

• 对于分页查询，返回记录的数量
• 可选地，如果返回记录数为零或与页面大小匹配，则返回查询产生的记录的总数，以计算总页数

这两个方面都不符合高效、非阻塞资源使用的概念。等待所有记录被接收（以确定分页详细信息的第一部分）将消除通过响应式数据访问获得的大部分好处。此外，执行计数查询相当昂贵，并增加了处理数据的延迟。

您仍然可以通过将Pageable (PageRequest)传递给存储库查询方法来自己获取数据块：

interface ReactivePersonRepository extends Repository<Person, Long> {

  Flux<Person> findByFirstnameOrderByLastname(String firstname, Pageable pageable);
}

Spring Data会通过将Pageable翻译为LIMIT和OFFSET来对查询应用分页。

参考文献：

• 参考文档：响应式仓库使用

- mp911de

---

你的示例中如何检索左侧页面？在非响应式样式中，存储库提供包含信息的Page<Person>类型。在使用Flux的示例中，Page信息不可用。 - zennon

你需要获取数量并从那里计算剩余的页面。 - mp911de

似乎Spring Data没有提供一个响应式的PageableHandlerMethodArgumentResolver来解析控制器中的Pageable。 - Hantsy

@mp911de 但是当我为分页编写测试时，我不确定为什么countByTitleLike的test在模拟环境中失败了。 - Hantsy

我不同意你的观点，从一个 REST 端点返回一组对象是非常常见的需求，即使它需要进行 100 次数据库请求，只要以非阻塞的方式进行就可以了，对吧？ - Johan Hendrik Ehlers

Answer 2

import com.thepracticaldeveloper.reactiveweb.domain.Quote;
import org.springframework.data.domain.Pageable;
import org.springframework.data.mongodb.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;

public interface QuoteMongoReactiveRepository extends ReactiveCrudRepository<Quote, String> {

    @Query("{ id: { $exists: true }}")
    Flux<Quote> retrieveAllQuotesPaged(final Pageable page);
}

更多细节，您可以在此处查看

- kn3l

---

+1 你的解决方案非常好用！！！我已经使用findAll和@Query("{ id: { $exists: true }}")更改了命名。 fun findAll(page: Pageable): Flux< Quote > - CRISTIAN ROMERO MATESANZ

Answer 3

我使用这种方法创建了一个服务，为那些仍在寻找解决方案的人提供帮助：

@Resource
private UserRepository userRepository; //Extends ReactiveSortingRepository<User, String>

public Mono<Page<User>> findAllUsersPaged(Pageable pageable) {
        return this.userRepository.count()
                .flatMap(userCount -> {
                    return this.userRepository.findAll(pageable.getSort())
                            .buffer(pageable.getPageSize(),(pageable.getPageNumber() + 1))
                            .elementAt(pageable.getPageNumber(), new ArrayList<>())
                            .map(users -> new PageImpl<User>(users, pageable, userCount));
                });
    }

- Sam1am

---

它不会在数据库中找到所有文档吗？ - inin

Answer 4

寻找响应式可分页存储库的一些想法时，我看到了一些解决方案，但它们会导致可怕的样板代码，所以我最终得出了这个（尚未在实际生活中尝试过，但应该可以正常工作，或者可能会成为您解决方案的灵感）。

因此...让我们创建一个全新的工具箱类，并使用此方法。

    public static 
           <R extends PageableForReactiveMongo<S, K>, S, T, K> Mono<Page<T>>
           pageableForReactiveMongo(Pageable pageable, 
                                             R repository, Class<T> clazzTo) {
        return repository.count()
                .flatMap(c ->
                        repository.findOderByLimitedTo(pageable.getSort(),
                                              pageable.getPageNumber() + 1)
                                .buffer(pageable.getPageSize(), (pageable.getPageNumber() + 1))
                                .elementAt(pageable.getPageNumber(), new ArrayList<>())
                                .map(r -> mapToPage(pageable, c, r, clazzTo))
                );
    }

而且它还需要类似这样的东西：

    private static <S, T> Page<T> mapToPage(Pageable pageable, Long userCount, Collection<S> collection, Class<T> clazzTo) {
        return new PageImpl<>(
                collection.stream()
                        .map(r -> mapper.map(r, clazzTo))
                        .collect(Collectors.toList())
                , pageable, userCount);
    }

然后，我们还需要一个抽象层来封装响应式存储库。

public interface PageableForReactiveMongo<D, K> extends ReactiveMongoRepository<D, K> {
    Flux<D> findOderByLimitedTo(Sort sort, int i);
}

让它由Spring实例化

@Repository
interface ControllerRepository extends PageableForReactiveMongo<ControllerDocument, String> {
}

最后像这样多次使用它

public Mono<Page<Controller>> findAllControllers(Pageable pageable) {
    return getFromPageableForReactiveMongo(pageable, controllerRepository, Controller.class);
}

这就是你的代码可能看起来的样子 :) 请告诉我它是否可以，或者对你有所帮助

- Michał Brzeziński

Answer 5

我使用了@kn3l的解决方案（不使用@Query）创建了另一种方法：

fun findByIdNotNull(page: Pageable): Flux< Quote>

它创建了相同的查询，而不使用@Query方法

- CRISTIAN ROMERO MATESANZ

Answer 6

我曾经遇到过同样的问题，最终采用了与上述类似的方法，但稍微更改了代码，因为我使用了Query DSL。以下是一个示例，如果有人需要。

@Repository
public interface PersonRepository extends ReactiveMongoRepository<Person, String>, ReactiveQuerydslPredicateExecutor<Person> {

    default Flux<Person> applyPagination(Flux<Person> persons, Pageable pageable) {
    return persons.buffer(pageable.getPageSize(), (pageable.getPageNumber() + 1))
        .elementAt(pageable.getPageNumber(), new ArrayList<>())
        .flatMapMany(Flux::fromIterable);
    }

}


public Flux<Person> findAll(Pageable pageable, Predicate predicate) {
    return personRepository.applyPagination(personRepository.findAll(predicate), pageable);
}

- user2669657

Answer 7

public Mono<Page<ChatUser>> findByChannelIdPageable(String channelId, Integer page, Integer size) {
    Pageable pageable = PageRequest.of(page, size, Sort.by(Sort.Direction.DESC, "chatChannels.joinedTime"));
    Criteria criteria = new Criteria("chatChannels.chatChannelId").is(channelId);
    Query query = new Query().with(pageable);
    query.addCriteria(criteria);
    Flux<ChatUser> chatUserFlux = reactiveMongoTemplate.find(query, ChatUser.class, "chatUser");
    Mono<Long> countMono = reactiveMongoTemplate.count(Query.of(query).limit(-1).skip(-1), ChatUser.class);
    return Mono.zip(chatUserFlux.collectList(),countMono).map(tuple2 -> {
        return PageableExecutionUtils.getPage(
                tuple2.getT1(),
                pageable,
                () -> tuple2.getT2());
    });
}

- yaoandw

---

感谢您的努力。请添加一些解释性文本描述您解决方案背后的思路。 - Markus-Hermann

仅包含代码的回答是低质量的回答。 - Taslim Oseni