Início / coding / Perguntas / 77935371
Accepted
Dawid
Asked: 2024-02-04 18:21:52 +0800 CST

Projeto Reactor qual é a diferença entre Mono.block() e Mono.subscribe()

  • 772

Não consigo encontrar a diferença no uso de Mono.block() e Mono.subscribe()

Para mim, ao usar os dois métodos, o código se comporta exatamente da mesma forma. e não deveria.

Para Mono.block() minha expectativa é que um thread que o chame bloqueie e aguarde resultados, mas ainda assim é usado no método map de um Mono e basicamente se desbloqueia.

Tenho o seguinte fragmento de código usando 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);
    }

Quando chamo esse código, recebo o seguinte:

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

Pelo que entendi, Mono.block() é um método de bloqueio, então presumo que o thread será bloqueado como seria ao adquirir um bloqueio. Mas em vez disso, o thread é usado para executar o código dentro do mapa do Mono, o que significa que ele não está bloqueado de forma alguma.

Para Mono.subscribe() eu esperaria que o thread que chama subscribe apenas continuasse e não esperasse pelos resultados, mas se comporta exatamente da mesma forma que quando usa Mono.block()

Eu tenho um fragmento semelhante, mas agora estou usando subscribe em vez de 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);
    }

Quando chamo esse código novamente, estou obtendo os mesmos resultados:

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

Eu esperaria que quando eu chamar subscribe, o thread atual continue funcionando, possivelmente exibindo:

On thread: [main] after subscribe
null

No meu caso, tanto o bloco quanto a assinatura se comportam exatamente da mesma forma, então qual é a verdadeira diferença?

spring-webflux

1 respostas

  1. Best Answer

    São suas suposições que estão erradas e seu teste não está correto para mostrar a diferença.

    nada acontece até você se inscrever (ou bloquear) .

    O Reactor é feito para abstrair o threading para você, para que você não precise usar mutex, atômicos, bloqueios, sincronizações, etc. Portanto, onde seu código é executado, em um thread, no mesmo thread, etc., não deve importar para você. O que importa é que ele usa threads NIO, o que significa que você nunca deve bloqueá-los.

    Pelo que entendi, Mono.block() é um método de bloqueio, então presumo que o thread será bloqueado como seria ao adquirir um bloqueio. Mas em vez disso, o thread é usado para executar o código dentro do mapa do Mono, o que significa que ele não está bloqueado de forma alguma.

    Aqui está sua primeira suposição: você acha que é um thread diferente que executará isso. Isso não é certo, pois o Reactor só usará threads diferentes quando necessário, caso contrário, ele apenas executará código, asynco que não significa necessariamente que será executado em threads diferentes. Você está confundindo a diferença entre asynce parallelporque eles não são a mesma coisa.

    Também bloqueia e mutexsó é usado se houver vários threads que desejam acessar o mesmo recurso, não há nenhum recurso aqui que queira ser acessado por vários threads.

    O que seu código de bloqueio está basicamente dizendo:

    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);
}

    E a saída do seu console mostra exatamente isso.

    Porém, quando se trata de assinatura, executamos o código, será um retorno de chamada, você basicamente assina algo que deve ser executado, mas não esperaremos pelos resultados, anexaremos um retorno de chamada ao resultado.

    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);
}

    Então você está errado ao presumir que sempre há threads diferentes em execução por baixo. Esse não é sempre o caso. Você pode forçar o reator a rodar em vários threads, mas se puder evitá-lo, ele o fará.

    Em segundo lugar, os computadores são rápidos, realmente muito rápidos, então você precisa introduzir atrasos para mostrar que a assinatura e o bloqueio são diferentes, e também dar ao reator um motivo para executar o código da maneira que você deseja.

    Na verdade, não posso escrever um exemplo agora porque estou no celular, mas encontrei isto que mostra exemplos:

    Qual é a diferença entre block() , subscribe() e subscribe(-)

    • 0

relate perguntas