主页 / coding / 问题 / 77935371
Accepted
Dawid
Asked: 2024-02-04 18:21:52 +0800 CST

Reactor 项目 Mono.block() 和 Mono.subscribe() 有什么区别

  • 772

我找不到使用 Mono.block() 和 Mono.subscribe() 的区别

对我来说，当使用这两种方法时，代码的行为完全相同。但它不应该。

对于 Mono.block() 我的期望是调用它的线程将阻塞并等待结果，但它在 Mono 的 map 方法中使用，并且基本上会自行解除阻塞。

我有以下使用 Mono.block() 的代码片段：

void doBlocking() {
        final var myMono = Mono.just("test").map(elem -> {
            System.out.printf("On thread: [%s] inside map\n",Thread.currentThread().getName());
            return elem;
        });


        String value;

        System.out.printf("On thread: [%s] before block\n",Thread.currentThread().getName());
        value = myMono.block();
        System.out.printf("On thread: [%s] after block\n",Thread.currentThread().getName());

        System.out.println(value);
    }

当我调用此代码时，我收到以下内容：

On thread: [main] before block
On thread: [main] inside map
On thread: [main] after block
test

根据我的理解 Mono.block() 是阻塞方法，所以我假设线程将像获取锁时一样被阻塞。相反，线程用于在Mono 的映射内部执行代码，这意味着它根本不会被阻塞。

对于Mono.subscribe()我希望调用 subscribe 的线程将继续而不等待结果，但它的行为与使用Mono.block()时完全相同

我有一个类似的片段，但现在使用订阅而不是块

void doSubscribing() {
        final var myMono = Mono.just("test").map(elem -> {
            System.out.printf("On thread: [%s] inside map\n",Thread.currentThread().getName());
            return elem;
        });


        AtomicReference<String> value = new AtomicReference<>();

        System.out.printf("On thread: [%s] before subscribe\n",Thread.currentThread().getName());
        myMono.subscribe(value::set);
        System.out.printf("On thread: [%s] after subscribe\n",Thread.currentThread().getName());

        System.out.println(value);
    }

当我再次调用此代码时，我得到相同的结果：

On thread: [main] before subscribe
On thread: [main] inside map
On thread: [main] after subscribe
test

我希望当我调用 subscribe 时，当前线程将继续工作，可能显示：

On thread: [main] after subscribe
null

就我而言，阻止和订阅的行为完全相同，那么真正的区别是什么？

spring-webflux

1 个回答

  1. Best Answer

    您的假设是错误的，并且您的测试不正确以显示差异。

    在您订阅（或阻止）之前什么都不会发生。

    Reactor 的目的是为您抽象出线程，这样您就不需要使用互斥体、原子、锁、同步等。这就是它的全部意义。因此，您的代码是在线程上、在同一线程上执行的，对您来说并不重要。重要的是它使用 NIO 线程，这意味着您永远不应该阻止它们。

    根据我的理解 Mono.block() 是阻塞方法，所以我假设线程将像获取锁时一样被阻塞。相反，线程用于在 Mono 的映射内部执行代码，这意味着它根本不会被阻塞。

    这是您的第一个假设，您认为它是一个不同的线程来运行它。这是不确定的，因为 Reactor 只会在需要时使用不同的线程，否则它只会运行代码async，这并不一定意味着它将在不同的线程上运行。您混淆了async和之间的区别parallel，因为它们不是同一件事。

    也是锁，mutex只有当有多个线程想要访问同一个资源时才使用，这里没有资源想要被多个线程访问。

    您的阻止代码基本上说的是：

    void doBlocking() {
    // I want to do this some time in the future
    final var myMono = Mono.just("test").map(elem -> {
        System.out.printf("On thread: [%s] inside map\n",Thread.currentThread().getName());
        return elem;
    });


    String value;

    System.out.printf("On thread: [%s] before block\n",Thread.currentThread().getName());

    // Here i want you to run the above code and we will wait here until it is done, 
    // this can be done on the same thread or a different thread. We dont care will most 
    // likely be the same thread for efficiency, because there is no reason to run a new 
    // thread for it.
    value = myMono.block();

    // Print the results
    System.out.printf("On thread: [%s] after block\n",Thread.currentThread().getName());

    System.out.println(value);
}

    你的控制台输出恰恰显示了这一点。

    不过，当谈到订阅时，我们运行代码，将回调，您基本上订阅了要执行的内容，但我们不会等待结果，我们会将回调附加到结果。

    void doSubscribing() {
    // Once again, declare we want to do this in the future sometime.
    final var myMono = Mono.just("test").map(elem -> {
        System.out.printf("On thread: [%s] inside map\n",Thread.currentThread().getName());
        return elem;
    });

    // Not really a need for Atomic reference because there is no guarantee 
    // that this will be executed on different threads its just you that
    // assumes so
    AtomicReference<String> value = new AtomicReference<>();

    System.out.printf("On thread: [%s] before subscribe\n",Thread.currentThread().getName());

    // Here the code is run with a callback, and since there are no delays 
    // the code is run so fast the callback is executed immediately, 
    // possibly on the same thread for efficiency, because switching threads
    // is resource demanding
    myMono.subscribe(value::set);

    // Here the result is printed
    System.out.printf("On thread: [%s] after subscribe\n",Thread.currentThread().getName());

    System.out.println(value);
}

    所以你假设总是有不同的线程在下面运行是错误的。情况并非总是如此。您可以强制reactor在多个线程上运行，但如果它可以避免它，它就会这样做。

    其次，计算机速度很快，就像非常非常快一样，因此您需要引入延迟来表明订阅和阻止是不同的，并且还给反应堆一个按照您想要的方式运行代码的理由。

    我现在无法真正编写示例，因为我在移动设备上，但我确实发现了这个显示示例：

    block() 、 subscribe() 和 subscribe(-) 之间有什么区别

    • 0

相关问题