Webclient 与简单 Flux.just 的 Flux 行为不同

Question

我在 Spring 中有基本的 rest 控制器。为了试用 spring webflux 并了解它的非阻塞性质。我创建了两个控制器映射，一个用于读取，一个用于服务 webclient 调用（如下所示）

@GetMapping("/slow-service-tweets")
private List<String> getAllTweets() {
    try {
        Thread.sleep(2000L); // delay
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return Arrays.asList(
            "Item1", "Item2","Item3");

}

这是我的测试get api，它只是触发下面给出的代码（第一个版本）

 @GetMapping("/test")
public void doSomething(){
    
   log.info("Starting NON-BLOCKING Controller!");
    Flux<String> tweetFlux = WebClient.create()
            .get()
            .uri("http://localhost:9090/slow-service-tweets")
            .retrieve()
            .bodyToFlux(String.class);

    tweetFlux.subscribe(tweet ->{
        try {
            log.info("i am back");
            Thread.sleep(6000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
       log.info(tweet.toString());});

  
   log.info("Exiting NON-BLOCKING Controller!");

以上代码的行为与它应该的完全一样。输出是

Starting NON-BLOCKING Controller!
Exiting NON-BLOCKING Controller!
Item1
Item2
Item3

原因是线程不会阻止订阅通量并继续前进。 现在请看下面的第二个版本。

@GetMapping("/test")
    public void doSomething(){
        
        System.out.println("i am here");

        Flux<Integer> f= Flux.just(1,2,3,4,5);
        //  Flux<Integer> f= Flux.fromIterable(testService.returnStaticList());



        f.subscribe(consumer->{
            try {
                log.info("consuming");
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info(consumer);
        });

        log.info("doing something else");
    }

最好像前面的例子 “做其他事情”必须立即打印。 但无论我做什么，打印所有元素都需要 10 秒，然后打印“做其他事情”。输出如下：

i am here
consuming
1
2
3
4
5
doing something else

谁能解释一下我在这里缺少什么？

Answer 1

我觉得我需要从一个警告开始 - 这肯定是不是如何使用 Webflux。任何时候你调用订阅，你几乎肯定做错了什么——这应该留给框架。相反，您应该使用：

• doOnNext() 用于日志记录等副作用；
• delayElements() 如果你想延迟 Flux 中的每个元素，而不是使用 Thread.sleep();
• 返回你的Flux（或返回一些通过使用运算符链接你的 Flux 创建的发布者）从你的 doSomething() 方法，以允许框架订阅并因此执行你的反应链。

如果您遵循上述“正常”的做事方式，您可能不会遇到这些阻塞/线程导致意外问题的问题。如果你做更多的小众事情，比如自己订阅、屏蔽线程而不考虑它们是否应该被屏蔽等等——那么你可能会给自己带来很多问题。

但是，要直接回答为什么会发生这种行为的问题 - 这是因为线程。当您使用Flux.just() 时，您使用的是即时调度程序（这意味着首先实际执行您的doSomething() 方法的线程。）因为这里只有一个线程在起作用，所以您的订阅方法会阻止这个线程它直到它完成，所以没有其他东西可以执行。例如，如果您告诉您的 Flux 在 boundedElastic() 线程池上发布，您会发现它的行为符合您的预期：

Flux<Integer> f = Flux.just(1, 2, 3, 4, 5).subscribeOn(Schedulers.boundedElastic());

在您的另一个示例中，当您使用 WebClient 时，它在与执行您的方法的线程不同的线程上发布 - 因此在订阅者中阻止这个不同的线程不会阻止您的 doSomething() 方法的整体执行。