filterWhen与flatMap结合在一起在Reactor中引发异常

Question

我想过滤所有值满足特定条件的窗口，然后打印出过滤后的通量：

Flux.just(4, 2, 6, 4, 5, 6, 7, 8, 9)
        .window(3)
        .filterWhen(window -> window.all(n -> n % 2 == 0))
        .flatMap(window -> window)
        .subscribe(System.out::println);

然而，结果不是4、2、6，而是

java.lang.IllegalStateException: UnicastProcessor allows only a single Subscriber.

如果我尝试相同但不使用过滤器，当它正常工作时：

Flux.just(4, 2, 6, 4, 5, 6, 7, 8, 9)
        .window(3)
        .flatMap(window -> window)
        .subscribe(System.out::println);

如果我在没有平面图的情况下尝试，也不会引发异常：

Flux.just(4, 2, 6, 4, 5, 6, 7, 8, 9)
        .window(3)
        .filterWhen(window -> window.all(n -> n % 2 == 0))
        .subscribe(System.out::println);

然而，这两种方法显然都没有产生预期的结果。但是，当和平面图时，在组合滤波器时似乎存在问题！

我的第一个例子有什么不对？另一个订阅者在哪里？

如何获取值？

Answer 1

这在目前是不可能的，因为一个窗口只能订阅（“消费”）一次。当将订阅窗口时，flatMap和filter都将订阅该窗口，因此会出现错误。

相反，您可以使用缓冲区（3）而不是窗口（3），过滤缓冲区，然后从通过过滤器的缓冲区中发出值：

Flux.just(4, 2, 6, 4, 5, 6, 7, 8, 9)
    .buffer(3)
    .filter(list -> list.stream().allMatch(n -> n % 2 == 0))
    .flatMapIterable(Function.identity())
    .subscribe(System.out::println);