什么是 Grouped Flux 以及我们如何使用它?

ayu*_*har 4 java spring reactive-programming project-reactor spring-webflux

我正在研究某个对象的 Flux 可以说Flux < MovieReservation >。这包含电影 id、名称、时间、标题等信息。所以我想提取有助于创建新
Flux < MovieShowDetail >. 我的意图是按电影 id 对所有保留进行分组,并将 Flux 分解为一组更小和多个 Flux(如果有的话,就是通量)。就像是

Flux {
   movie1 -> Flux<MovieShowDetail>
   movie2 -> Flux<MovieShowDetail>
   ... and so on
}
Run Code Online (Sandbox Code Playgroud)

所以我遇到了这个 groupBy 方法,它应该只做这样的事情。然而,文档确实没有内容,特别是关于如何迭代每部电影及其各自的 Flux。

此外,当我尝试通过尝试和错误来学习时,处理在 groupBy 方法之前的操作后停止。

我试过做

fluxOfSomething
.groupBy( movieReservation -> movieReservation.getMovieId ,
movieReservation -> movieReservation) 
Run Code Online (Sandbox Code Playgroud)

这样我就可以遍历每个通量并创建新的 MovieShowDetail 通量。但是,处理永远不会进入这个块。我尝试记录东西,但流程从未进入过它。

flux
    .map( movieSomething -> do something)
    .groupBy( movieReservation -> 
        movieReservation.getMovieId , movieReservation ->
    movieReservation)
    .subscribe("This text doesn't get printed");
Run Code Online (Sandbox Code Playgroud)

我真的需要尽可能多的信息。

Sim*_*slé 6

groupBy产生 a Flux<Flux<T>>(或更准确地说 a Flux<GroupedFlux<T>>,它公开每个组的键)。

AGroupedFlux和 a 一样Flux,必须订阅才能变为活动状态。所以,你需要做的是某种消费内FluxgroupBy产生。

这样做的一种典型方法是使用flatMap,它已经进行了Function<T, Flux>转换。该函数可以很简单Function.identity()(但如果您想进一步处理内部的每个元素,Flux您可能应该从内部进行处理flatMap Function(因为组键在该 lambda 的范围内)。

movieReservations
    .groupBy(MovieReservation::movieId)
    .flatMap(idFlux -> idFlux
        .collectList()
        .map(listOfReservations ->
            new MovieInformation(idFlux.key(), listOfReservations)
        )
    );
Run Code Online (Sandbox Code Playgroud)