java - react 堆项目 : downstream is slow

Question

据我了解，下游的其余部分需要在线程池中的线程上处理(我将其设置为1024)

这是我的代码。

Flux<String> ips =
        Flux.fromIterable(items).map(Item::getIp);
ips
        .publishOn(Schedulers.fromExecutor(Executors.newFixedThreadPool(1024)))
        .map(ip -> {
            try {
                Request request = new Request.Builder().url("https://" + ip + ":443").build();
                Response response = okHttpClient.newCall(request).execute();
                return response.code();
            } catch (Exception e) {
            }

            return -1;
        })
        .subscribe(System.out::println);

出于某种原因，与以下代码相比，此代码非常慢:

appRules
        .stream()
        .parallel()
        .map(Item::getIp)
        .forEach(ip -> {
            try {
                Request request = new Request.Builder().url("https://" + ip + ":443").build();
                Response response = okHttpClient.newCall(request).execute();
                System.out.println(response.code());
            } catch (Exception e) {
            }

            System.out.println(-1);
        });

为什么？当您受到 IO 限制时，同时处理项目流的正确方法是什么？ (而不是CPU)

Answer 1

执行速度较慢的原因是Reactor管道执行默认是单线程的。因此，当您使用 Flux.publishOn 运算符时，您只需说您希望在给定线程池中的线程上执行管道的这一部分，但它不会同时在单独的线程上执行每个项目。

实现并发的一种选择是使用 parallel Flux ，它创建了所谓的“rails”，其中数据可以并行流动，但它主要用于 CPU 密集型操作。

更好的选择是将阻塞代码包装在 Mono 中并将其委托(delegate)给专用线程池，类似于您所做的，只是这次每个任务都会获得自己的线程:

private static void reactorProcess()
{
    ExecutorService executor = Executors.newFixedThreadPool(1024);

    Flux.range(1, 1024)
        .flatMap(a -> Mono.fromRunnable(() -> simulateHttpCall())
                          .subscribeOn(Schedulers.fromExecutor(executor)))
        .blockLast();

    executor.shutdown();
}

private static void simulateHttpCall()
{
    try
    {
        Thread.sleep(100);
        System.out.println(Thread.currentThread().getName() + ": " + ZonedDateTime.now());
    } catch (InterruptedException e)
    {
        e.printStackTrace();
    }
}

我还要指出，Java 并行流并不是这种处理的可行替代方案。它使用 ForkJoinPool默认情况下，它也适用于 CPU 密集型操作，并且仅使用与计算机中 CPU 核心数量相同的线程。

除此之外，如果您想充分利用响应式(Reactive)编程的功能，您应该考虑使用支持非阻塞 IO 的 HTTP 客户端，例如 WebClient从 Spring 开始。通过使用非阻塞 HTTP 客户端，您无需再担心定义线程池，因为不会阻塞任何线程，并且固定的少量线程将能够服务数千个并发请求。