使用 Reactor 对输入通量进行乱序处理

Question

我试图弄清楚如何获取作为输入数据序列的 Flux，并行通过可能重新排序序列的阻塞调用运行它们，然后通过第二个单线程阻塞调用。这个想法是最终的单线程调用将重新排序的并行工作输出记录到磁盘上。我试图做的最终目标是并行算法是一种共识算法，它将确定数据输入的实际顺序。单线程写入是模拟按照共识算法确定的顺序处理事物。

查看this article，它建议我应该将阻塞调用转换为在调度程序上运行的 Mono，该调度程序可以为我提供并行或单线程处理：

public class BlockingRemoteCall {

    private final static Random r = new Random();

    private final static Scheduler scheduler = Schedulers.newParallel("myWebservice", 10);

    static private String blockingWebService(final String in) {
        try {
            // fakes blocking for up to a second
            Thread.sleep((long) (1000 * r.nextFloat()));
            System.out.println("webserver returned: "+in+" on "+Thread.currentThread().getName());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return in;
    }

    public static Mono<String> blockingMethodParallelThread(final String in) {
        return Mono.fromSupplier(() -> blockingWebService(in))
                .subscribeOn(scheduler);
    }
}

public class BlockingJournal {

    private final static Scheduler scheduler = Schedulers.newSingle("myJournal");

    private static String blockingWrite(String in){
        try {
            // fakes blocking for disk write
            Thread.sleep(5L);
            System.out.println("journal wrote: "+in+" on "+Thread.currentThread().getName());
        } catch (Exception e){
            throw new RuntimeException(e);
        }
        return in;
    }

    public static Mono<String> blockingMethodSingleThread(final String in) {
        return Mono.fromSupplier(() -> blockingWrite(in))
                .subscribeOn(scheduler);
    }
}

我一直在尝试通过这些方法获取整数的 Flux 并以某种方式映射或 flatMap，但我无法获取任何要记录的内容。这是我最近的尝试：

        final Scheduler parallelScheduler = Schedulers.newParallel("p");
        final Scheduler singleScheduler = Schedulers.single();

        Flux<String> flux = Flux.range(1, 10).map(i -> i.toString()).publishOn(parallelScheduler);

        Flux<String> pipeline = flux.map(s->{
            Mono<String> async = BlockingRemoteCall.blockingMethodParallelThread(s);
            String r1 = async.block();
            Mono<String> r2 = BlockingJournal.blockingMethodSingleThread(r1);
            return r2.block();
        });

        pipeline.subscribeOn(singleScheduler).doOnNext(System.out::println).blockLast();

这实际上并没有输出任何东西，但每当我能够生成任何输出时，我只看到println 语句显示数据流是在一个线程上按顺序处理的。我希望看到的是，对blockingMethodParallelThread(s) 的调用中的任意延迟导致输入序列无序记录。

我该如何设置，以便可以并行处理输入通量（最终从反应器网络输入冒泡），重新排序，然后最终按顺序处理，保留重新排序？在哪里重新排序是由于并行阻塞调用？谢谢！

Answer 1

这里有几点：

• 仅仅调用publishOn(parallelScheduler) 不会让你的Flux 并行执行，这只是意味着你的顺序Flux 现在将在并行调度程序上发布。相反，您可能希望调用parallel() 使其并行，然后使用您选择的调度程序指定runOn()。 （类似地，在并行的Flux 上调用sequential() 将使其再次连续。）
• 您几乎肯定不想在这项工作中使用并行调度程序 - 有界弹性调度程序会是更好的选择（它是专门为包装阻塞 IO 而设计的。）
• 进行map() 调用然后在其中阻塞没有多大意义 - 您不妨改用flatMap() 并返回结果发布者。
• 在这里创建新的调度器没有任何好处，你也可以使用默认的。

因此，考虑到这些要点，您的代码变为：

ParallelFlux<String> flux = Flux.range(1, 10).map(i -> i.toString()).parallel().runOn(Schedulers.elastic());

ParallelFlux<String> pipeline = flux.flatMap(s -> {
    Mono<String> async = BlockingRemoteCall.blockingMethodParallelThread(s);
    String r1 = async.block();
    return BlockingJournal.blockingMethodSingleThread(r1);
});

pipeline.sequential().doOnNext(System.out::println).blockLast();

...这将按照您的预期无序输出结果。