什么是 Grouped Flux,我们究竟如何使用它?

What is a Grouped Flux and how exactly do we work with one?

我正在研究一些 object 的 Flux 让我们说 Flux < MovieReservation > 。这包含电影 ID、名称、时间、标题等信息。所以我想提取有助于创建新
Flux < MovieShowDetail > 的信息。我的意图是按电影 ID 对所有预订进行分组,并将 Flux 分解为一组更小的多个 Flux(如果有的话,则为 flux)。像

Flux {
   movie1 -> Flux<MovieShowDetail>
   movie2 -> Flux<MovieShowDetail>
   ... and so on
}

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

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

我试过了

fluxOfSomething
.groupBy( movieReservation -> movieReservation.getMovieId ,
movieReservation -> movieReservation) 

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

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

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

groupBy 生成一个 Flux<Flux<T>>(或者更准确地说是一个 Flux<GroupedFlux<T>>,它公开了每个组的密钥)。

A GroupedFlux,如 Flux,必须订阅才能激活。所以你需要做的是以某种方式消耗 groupBy 产生的内部 Flux

一种典型的方法是使用 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)
        )
    );