Spring WebFlux：flatmap异步转换

Question

我已经了解到flatmap转换是异步的，在这个示例中，我在lambda定义中打印了线程的名称。它正在打印订阅源的同一线程。根据我的理解，它应该打印一个不同的线程名--而不是订阅源的线程名，因为这个转换必须在不同的线程中执行。

Flux.just(1, -2, 3, 4, -5, 6)
    .flatMap(element -> { 
        try { 
            Thread.sleep(1000);
        } 
        catch (InterruptedException e) {
            e.printStackTrace(); 
        } 

        System.out.println(Thread.currentThread().getName() + " element: " + element); 
        return Flux.just(element);
    })
    .subscribe()

Answer 1

它是异步的这一事实并不一定意味着它是并行运行的，这似乎是您在这里期待的。但是，您可以将flux转换为Parallelflux并指定并行调度程序：

Flux.just(1, -2, 3, 4, -5, 6)
        .parallel()
        .runOn(Schedulers.elastic())
        .flatMap(element
                -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) { // TODO Auto-generated catch block
                e.printStackTrace();
            }

            System.out.println(Thread.currentThread().getName()
                    + " element: " + element);

            return Flux.just(element);
        })
        .subscribe();
Thread.currentThread().join(); //Just a hack to keep the program alive.

另一方面，如果您不希望它并行运行，而只是在主线程的一个单独线程上运行，则无需将其转换为并行流量-只需提供.subscribeon（schedulers.ellastic())调用或类似的调用即可。