反应堆-异步/非阻塞

Question

我认为下面的磁通链将通过事件循环(像JS)被放置/执行。因此，运行下面的代码将首先打印阻塞的for循环&然后执行通量链。

但是，整个通量总是在移动到for循环之前先执行。我确实有一些sleep语句是阻塞的。但是有两个doOnNext阶段

AtomicInteger atomicInteger = new AtomicInteger(0);

// reactor
Flux.generate(synchronousSink -> {
            if (atomicInteger.incrementAndGet() < 3) {
                synchronousSink.next(atomicInteger.get());
            } else
                synchronousSink.complete();
    })
    .doOnNext(i -> {
        System.out.println(
                "A - Received " + i + " by " + Thread.currentThread().getName()
        );
        sleep(Duration.ofSeconds(1));
    }).doOnNext(i -> {
        System.out.println(
                "B - Received " + i + " : by " + Thread.currentThread().getName()
        );
        sleep(Duration.ofSeconds(1));
    }).subscribe();


for (int i = 0; i < 5; i++) {
    System.out.println("For " + i + " by " + Thread.currentThread().getName());
    sleep(Duration.ofMillis(500));
}

印出来

A - Received 1 by main
B - Received 1 by main
A - Received 2 by main
B - Received 2 by main
For 0 by main
For 1 by main
For 2 by main
For 3 by main
For 4 by main

有人能解释一下这种行为并回答这些问题吗？

当我们使用反应器时，只有使用一些调度器才能实现异步/非阻塞行为?如果我不使用任何调度器，并且让代码使用当前线程执行，那么即使对于IO密集型的applications?，我们也可以使用WebFlux而不是Spring来实现更好的性能差异吗？

Answer 1

1. 线程阻塞是反应堆使用不当的一种方式。要使它以非阻塞的方式工作，您应该使用publishOn/subscribeOn，然后输出应该是：

。

主A接收

0-有界弹性接收1，主B接收1，主B接收2，主A接收3，主A接收4，有界弹性接收2

有关publishOn vs subscribeOn的更多信息，请参见：link

1. 为确定反应堆支持非阻塞的HTTP (包括Websockets)，TCP和UDP.此外，默认情况下，反应堆在Netty服务器上工作，这改变了处理请求的方式。例如，在Tomcat请求中--响应由同一个线程处理--而且这个线程正在等待响应，所以它被阻塞了。在Netty中，一个线程可以处理发送请求，另一个线程可以处理接收响应-线程不等待响应。