.map() 函数后的通量流不按顺序排列答案

【问题标题】：Flux stream after .map() function not in order.map() 函数后的通量流不按顺序排列
【发布时间】：2021-11-21 16:15:03
【问题描述】：

在尝试熟悉 Flux 时，我希望在 Flux 中有一个数据流。

.map() 函数应该模拟一个工作负载，每个Magician 对象需要一些随机处理时间，并且发生在单独的线程中（正如您在Thread id 的输出中看到的那样，以利用多线程。

最后，流应该按顺序返回结果。但无论我做什么，例如使用.sequential()，或者它总是以无序结束。

有人可以告诉我如何使用多线程（使用.map() 或任何其他序列，只要结果是所要求的）并得到结果按顺序好吗？谢谢

附：这个问题背后的业务逻辑是我正在调用一个提供Flux<byte[]>的端点

我需要并行处理（解密）这些字节，并因此转发给另一个消费者。

public class FluxRunner {

    public void getWorkerStream() throws InterruptedException {
        final int[] counter = new int[1];

        // create workers
        final Flux<Magician[]> workersFlux = Flux.range(0, 10)
                .map(integer -> {
                    final Magician[] worker = new Magician[1];
                    worker[0] = new Magician(counter[0], counter[0]++);
                    return worker;
                });

        final Disposable disposable = workersFlux
                .parallel()
                .runOn(Schedulers.parallel())
                .map(workers -> {
                    System.out.println("Thread id: " + Thread.currentThread().getId());
                    workers[0].calculate();
                    return workers;
                })
                .sequential() // no effect, doOnNext is unordered
                .doOnNext(workers -> System.out.println(workers[0].getId()))
                .subscribe();

        while (!disposable.isDisposed()) {
            sleep(500);
        }
    }
}

@Data
class Magician {

    final int id;
    int number;

    Magician(final int id, final int number) {
        this.id = id;
        this.number = number;
    }

    public void calculate() {
        int timeToSleep = (int) (Math.random() * 3000);
        System.out.println("Sleep for " + timeToSleep + " seconds");
        try {
            sleep(timeToSleep);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        number++;
    }
}

结果将是

Thread id: 33
Thread id: 24
Thread id: 25
Thread id: 31
Thread id: 29
Thread id: 30
Thread id: 27
Thread id: 32
Thread id: 28
Thread id: 26
Sleep for 2861 seconds
Sleep for 2811 seconds
Sleep for 711 seconds
Sleep for 2462 seconds
Sleep for 1858 seconds
Sleep for 601 seconds
Sleep for 126 seconds
Sleep for 359 seconds
Sleep for 2014 seconds
Sleep for 2356 seconds
4
5
7
9
8
3
0
1
6
2

【问题讨论】：

除非您使用类似于sorted() 运算符的东西，否则我不知道有一种方法可以订购由parallel() 拆分的流。如果确实存在某种方式，您确实意识到您可能会超越与您的要求平行获得的收益，对吗？基本上所有完成的任务都需要等待第一个任务发出，然后是第二个任务，等等......

标签： java spring-webflux reactor

【解决方案1】：

您可以使用flatMapSequential 运算符和.subscribeOn(Schedulers.parallel()) 来获得所需的结果：

final Disposable disposable = workersFlux
    .flatMapSequential(workers -> Mono.fromCallable(() -> {
        System.out.println("Thread id: " + Thread.currentThread().getId());
        workers[0].calculate();
        return workers;
    }).subscribeOn(Schedulers.parallel()))
    .doOnNext(workers -> System.out.println(workers[0].getId()))
    .subscribe();

【讨论】：