项目Reactor-如何通过窗口处理结果

Question

Flux.range(0, 100)
    .log("before window")
    .window(10)
    .map(Flux::toList)
    .log("after window")
    .map((w) -> {
        System.out.println(w.subscribe().get()));
        return 1;
    }) 
    .reduce(0, (a, b) -> a + b)
    .doOnSuccess(System.out::println)
    .subscribe();

[           main] before window : onNext(0)
[           main] after window  : onNext({ operator : "BufferAll" })

官方文档中没有那么多信息，所以我想我误解了什么，并且错误地使用了窗口函数。但我到底做错了什么？

更新

我发现，如果使用.doonsuccess，就可以避免这个特殊的问题。比如：

.map((w) -> {
    w.doOnSuccess((w2) -> System.out.println(w2))).subscribe();
    return 1;
}

但真正的问题是，在我的情况下，我需要根据提供的数据返回一个数字（而不是1)。我可以在这里创建一个新的，但无论如何，以后我应该.get。例如，在final.reduce中。因此，如果我执行.reduce(0,(a,b)->a+b.get())，那么它将失败。

.window(10)
.log("after window")
.map((w) -> {
    //basically i'm reducing Flux to a Mono<List> and return number of a [good] elements in it
    return w.reduce(...).map(ids -> 100).subscribe();
})
.reduce(0, (a, b) -> a + b.get()) 

但它无论如何都不起作用，困在.reduce:

注意到如果删除.reduce步骤，它就可以工作了。在这种情况下，.window提供的flux的处理在main flow之后执行。我对此没有任何控制权，甚至无法得到最终结果。这没有任何意义。

Answer 1

这个问题是因为我需要在进一步使用之前减少窗口。

比如：

window(...).flatMap( (window) -> window.reduce(...))

我是在映射器的单声道中这样做的，但它在那里阻止了执行流，所以它不是一个正确的位置。它必须在窗口之后，下次使用之前。

Flux.range(0, 100)
    .window(10)
    .flatMap(window -> {
         return window.reduce(new ArrayList<>(), (a, b) -> {
             a.add(b);
             return a;
         });
    })
    .map((list) -> list.size())
    .reduce(0, (a, b) -> a + b)
    .doOnSuccess(System.out::println)
    .subscribe();