vertx executeBlocking другое поведение

Question

Я экспериментировал с vertx executeBlocking, чтобы смоделировать сценарий реального времени, который я сделал после

vertx.setPeriodic(1000, id ->{
    counter += 1;
    LOGGER.info("invoked method {} ",counter);
    vertx.executeBlocking(future -> {
        int counterFinal = counter;
        String result = service.blockingMethod("cycle "+counterFinal+" executed");
        future.complete(result);
    }, res -> {
        LOGGER.info(String.format("The result is: %s", res.result()));
    });

метод блокировки довольно прост

public String blockingMethod(String result){
    block(2);
    return result;
}

И это был результат

07:50:27.742 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 1 
07:50:28.742 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 2 
07:50:29.740 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 3 
07:50:29.764 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - The result is: cycle 1 executed
07:50:30.739 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 4 
07:50:31.739 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 5 
07:50:31.773 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - The result is: cycle 3 executed
07:50:32.751 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 6 
07:50:33.748 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - invoked method 7 
07:50:33.789 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.AsyncExperimentalVerticle - The result is: cycle 5 executed

явно отсутствует два события в среднем, потому что задержка была установлена ​​на 2 секунды.Затем я обернул блокирующий метод внутри класса и затем выполнил следующим образом

vertx.setPeriodic(1000, id ->{
    counter++;
    LOGGER.info("invoked method {} ",counter);
    service.wrapperMethod("Hello", counter, new Handler<AsyncClass>() {
        @Override
        public void handle(AsyncClass event) {
            vertx.executeBlocking(future -> {
                String result = event.result();
                future.complete(result);
            }, res -> {
                LOGGER.info(String.format("The result is: %s", res.result()));
            });
        }
    });
});

, а метод-обёртка был спроектирован таким образом

public void wrapperMethod(String input, int cycle, Handler<AsyncClass> execute) {
    AsyncClass instance = new AsyncClass(input,String.valueOf(cycle)); // my custom class where the result method has a 2 sec delay
    execute.handle(instance);
}

, тогда я получил ожидаемый результат.

08:08:27.358 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 1 
08:08:27.368 [vert.x-worker-thread-0] INFO lab.async.base.support.AsyncClass - Invoking method inside AsyncClass class
08:08:28.338 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 2 
08:08:29.345 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 3 
08:08:29.384 [vert.x-worker-thread-0] INFO lab.async.base.support.AsyncClass - Invoking method inside AsyncClass class
08:08:29.386 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - The result is: Hello world of cycle 1
08:08:30.347 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 4 
08:08:31.351 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 5 
08:08:31.391 [vert.x-worker-thread-0] INFO lab.async.base.support.AsyncClass - Invoking method inside AsyncClass class
08:08:31.391 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - The result is: Hello world of cycle 2
08:08:32.341 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 6 
08:08:33.343 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - invoked method 7 
08:08:33.396 [vert.x-worker-thread-0] INFO lab.async.base.support.AsyncClass - Invoking method inside AsyncClass class
08:08:33.397 [vert.x-eventloop-thread-0] INFO lab.async.base.verticle.TestVerticle2 - The result is: Hello world of cycle 3

Теперь я вижу асинхронное выполнение без пропуска ни одного события.Я не могу найти возможное объяснение.Даже в методе-обертке, если я задержу n секунд, он пропустит события, как и ожидалось.

Кто-то, пожалуйста, помогите мне разобраться в этом поведении.

update1:

за секундуСценарий структуры AsyncClass приведен ниже

public class AsyncClass {

    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncClass.class);

    private String input;
    private String cycle;

    public AsyncClass(String input, String cycle) {
        this.input = input;
        this.cycle = cycle;
    }

    public String result(){
        LOGGER.info("Invoking method inside AsyncClass class");
        block(2);
        return input+" world of cycle "+cycle;
    }

    private void block(int pauseLimitInSecond){
        try {
            TimeUnit.SECONDS.sleep(pauseLimitInSecond);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOGGER.error("exception - > ", e);
        }
    }
}

Answer 1

Наблюдаемое поведение обусловлено Перехватом лямбда-переменной тела , где захватывается ссылка на внешний класс (this).

Переменная экземпляра внешнего класса *Значение 1008 * будет изменено, когда будет вычислено выражение Lambda body (увеличено на 1 по сравнению с начальным значением), что создает иллюзию неожиданного поведения.

Сохранение того жеВ последовательности программы вы можете заменить тело лямбда-выражения реализацией Handler<Future<String>>, где внешняя переменная экземпляра counter хранится в другой переменной экземпляра, которая будет использоваться в теле выполнения обработчика:

private static final class BlockingHandler implements Handler<Future<String>> {

    private final YourBlockingService service;

    private final int counter;

    public BlockingHandler(int counter, YourBlockingService service) {
        this.counter = counter;
        this.service = service;
    }

    @Override
    public void handle(Future<String> event) {
        String result = service.blockingMethod("cycle " + this.counter + " executed", 2);
        event.complete(result);
    }
}

Ваш код вертикали будет выглядеть следующим образом:

this.vertx.setPeriodic(
      1000,
      id -> {
          counter += 1;
          LOGGER.info("invoked method {}", counter);
          vertx.executeBlocking(
                new YourBlockingHandler(this.counter, this.service),
                res -> LOGGER.info(String.format("The result is: %s", res.result()))
          );
      }
);

В заключение, упомянутое поведение относится только к семантике замыкания и не имеет ничего общего с Vert.x внутренности.

Answer 2

Используемый вами метод executeBlocking обеспечивает выполнение задач блокировки по порядку (одна за другой).

То есть при выполнении второй задачи блокировки (через две секунды)переменная counter уже была увеличена в два раза.Следовательно, серии 1, 3, 5.

В вашей другой попытке с классом переноса значение переменной counter захватывается до вызова executeBlocking.Поэтому при выполнении задачи блокировки вы получите ожидаемое значение.