什么是分组通量，我们如何使用它？

Question

我正在研究一些对象的Flux，比如Flux

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

所以我遇到了这个groupBy方法，它应该只做这样的事情。然而，关于这方面的文档实在是太多了，尤其是关于如何迭代每部电影及其各自的流量。

此外，当我尝试通过尝试和错误来学习时，处理会在处理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");

我真的需要尽可能多的信息，你可以提供关于这一点。




Answer 1

groupBy产生通量

GroupeFlux，如Flux，必须订阅才能激活。因此，您需要做的是以某种方式消耗groupBy产生的内部Flux。

一种典型的方法是使用flatMap，它已经具有了一个函数

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