使用Flux会丢失一些物品。Reactor中的push（）

Question

我的代码如下：

public class SequenceCreator {

    public Consumer<List<Integer>> consumer;

    public Flux<Integer> createNumberSequence() {
        return Flux.push(sink -> consumer = items -> items.forEach(sink::next));
    }

    public static void main(String[] args) throws InterruptedException {
        SequenceCreator sequenceCreator = new SequenceCreator();

        List<Integer> sequence1 = Lists.newArrayList(1,2,3,4,5);
        List<Integer> sequence2 = Lists.newArrayList(6,7,8,9,10);

        Thread producingThread1 = new Thread(
                () -> sequenceCreator.consumer.accept(sequence1));

        Thread producingThread2 = new Thread(
                () -> sequenceCreator.consumer.accept(sequence2));

        sequenceCreator.createNumberSequence().subscribe(System.out::println);

        producingThread1.start();
        producingThread2.start();

        while (true) {
            Thread.sleep(1000);
        }
    }
}

输出为

1 2 3 4 5 7 8 9 10

我不知道为什么数字6没有输出这是多线程的原因吗？

Answer 1

我不知道为什么数字6没有输出这是多线程的原因吗？

是的，几乎可以肯定。看一下Javadoc forFlux。推送：

以编程方式创建Flux，该Flux能够通过FLuxSink API从单线程生产者发出多个元素。有关支持多线程的替代方案，请参阅创建（消费者）。

您没有使用单线程生产者（违反记录的要求），因此在这个场景中行为基本上是未定义的。您需要切换到通量。按照文档的建议创建，因为您使用多个线程来发布。