无功Flux编程中的并行执行

Question

我试图避免使用阻塞线程模型，而使用反应式模型来实现高吞吐量。我的用例是这样的：有大量的传入消息。对于每条消息，我需要在不阻塞该线程的情况下执行一些I/O操作。在这里，我在一个单独的线程中处理每条消息。如果应用程序被终止，我需要完成正在进行的任务并优雅地关闭。我用的是线。在下面睡眠以模拟密集的I/O操作。代码示例如下：

public class TestReactor {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        Disposable task = Flux.range(1, 100).parallel().runOn(Schedulers.fromExecutor(executor)).doOnNext(message -> {
            System.out.println(Thread.currentThread().getName()+": processing " + message);
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+": Done");
        })
        .sequential()
        .doOnError(e->e.printStackTrace())
        .doOnCancel(()->{
            System.out.println("disposing.....");
            executor.shutdown();
            try {
                executor.awaitTermination(5, TimeUnit.SECONDS);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        })
        .subscribe();

        Thread.sleep(4000);
        
        task.dispose();
        
        System.out.println("Disposed. Waiting for sometime before exit.");
        Thread.sleep(20000);
    }

}

当我运行这个时，通量似乎忽略了执行者。shutdown（）和错误以及中断的异常。是否可以使用flux实现我的用例？

Answer 1

您犯了一个很大的错误：在使用反应式编程时，永远不要使用任何线程操作。这是一个肮脏的黑客。当您尝试在FRP上编码时，最明显的不良设计和代码气味标记是：

• 调用操作符和副作用函数（doOn）内的try catch块。如果你有一个例外情况-好的，没关系。调用onError操作符并在其中处理管道的行为

所以，关于这个问题及其解决方案：你做了糟糕的设计：）你不需要处理任何一次性的或线程中断。您只需删除它并添加killswitch。

那么，工作代码，这就是您想要的：

public class TestReactor {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        Flux.generate(() -> 0,
                (state, sink) -> {
                    sink.next("current state = " + state);
                    if (state == 100) sink.complete();
                    return state + 1;
                }
        )
        .parallel()
        .runOn(Schedulers.fromExecutor(executor))
        .doOnNext(message -> {
            System.out.println(Thread.currentThread().getName() + ": processing " + message);
            System.out.println(Thread.currentThread().getName() + ": Done");
        })
        .sequential()
        .doOnError(e -> e.printStackTrace())
        .doOnCancel(() -> {
            System.out.println("disposing.....");
        })
        .subscribe();
    }

}