项目反应堆:collectList()对Flux.create()不起作用 [英] Project reactor: collectList() doesn't work for Flux.create()

问题描述

下面的示例显示从1到10的整数以及(7，8，9，10)的列表

Below example prints integers from 1 to 10 and a list of (7, 8, 9, 10)

public void streamCollect() {

    ConnectableFlux<Integer> connect = Flux.range(1, 10)
            .publish();

    connect.subscribe(v -> System.out.println("1: " + v));

    connect
            .filter(number -> number > 6)
            .collectList()
            .subscribe(v -> System.out.println("4: " + v));

    connect.connect();
}

结果:

1:1

1:2

1:3

1:4

1:5

1:6

1:7

1:8

1:9

1:10

4:[7、8、9、10]

4: [7, 8, 9, 10]

下一个示例应产生相同的结果，但只输出1到10之间的数字，而不打印列表.为什么?

Next example should produce the same result but instead prints out only numbers from 1 to 10 but no list. Why?

public void streamCollect() {

    ConnectableFlux<Integer> connect = Flux.<Integer>create(emitter -> {

        Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
                .forEach(t -> emitter.next(t));
    }).publish();

    connect.subscribe(v -> System.out.println("1: " + v));

    connect
            .filter(number -> number > 6)
            .collectList()
            .subscribe(v -> System.out.println("4: " + v));

    connect.connect();
}

结果:

1:1

1:2

1:3

1:4

1:5

1:6

1:7

1:8

1:9

1:10

推荐答案

collectList等待onComplete信号，您从未在创建lambda中产生该信号

The collectList waits for the onComplete signal, which you never produce in your create lambda

这篇关于项目反应堆:collectList()对Flux.create()不起作用的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！