为什么同时从多个线程调用 Vert.x worker verticle？

Question

我用 Java (11) 编写的 vertx (4.0.2) 应用程序使用了一些数据量大的 Verticle，这些 Verticle 会导致延迟峰值，因为事件循环会被它们暂时阻塞。出于这个原因，我想将这些verticles部署为worker verticles，这样eventloop和其他verticles就不再被阻塞了。

不幸的是，我的应用程序现在崩溃了，因为 Verticle 内的事件处理是由多个线程同时执行的；(

如果我正确理解了 vertx 文档，则不会发生这种情况：

Vert.x 从不会由多个线程同时执行 Worker Verticle 实例，但可以在不同时间由不同线程执行。

我能够通过一个最小的示例重现该问题：

@Slf4j
public class WorkerTest extends AbstractVerticle {
  private static final String ADDRESS = "address";
  private volatile String currentThread = null;
  private long counter = 0;

  @Override
  public void start(final Promise<Void> startPromise) {
    vertx.eventBus().consumer(ADDRESS, this::handleMessage);
    startPromise.complete();
  }

  private void handleMessage(Message<Object> message) {
    final var _currentThread = this.currentThread;
    final var thisThread = Thread.currentThread().getName();

    if (_currentThread != null) {
      log.error(
          "concurrent callback: current thread={}, this thread={}", _currentThread, thisThread);
      return;
    }

    try {
      this.currentThread = thisThread;
      Thread.sleep(2);
      if (++counter % 100L == 0) {
        log.info("received {} messages (current thread: {})", counter, thisThread);
      }
    } catch (Exception e) {
    } finally {
      this.currentThread = null;
    }
  }

  public static void main(String[] args) {
    final Vertx vertx = Vertx.vertx();

    vertx.deployVerticle(
        new WorkerTest(),
        new DeploymentOptions().setWorker(true),
        result -> {
          if (result.failed()) {
            System.exit(1);
            return;
          }

          for (int i = 0; i < 1000; ++i) {
            vertx.eventBus().send(ADDRESS, "test");
          }
        });
  }
}

执行这会给我很多日志错误，因为handleMessage 是同时从多个线程调用的。如果我将 Verticle 部署为非工作人员，这将按预期工作。

我在这里做错了什么？

Answer 1

vertx 4.0.2 似乎是您的问题。使用 vertx 4.0.3 和以下代码：

public class WorkerTest extends AbstractVerticle {
    private static final String ADDRESS = "address";

    private volatile boolean handleMessageInExecution = false;

    public static void main(String[] args) {
        final Vertx vertx = Vertx.vertx();

        vertx.deployVerticle(
                WorkerTest::new,
                new DeploymentOptions()
                        .setInstances(2)
                        .setWorkerPoolSize(10)
                        .setWorker(true)
                ,
                result -> {
                    for (int i = 0; i < 100; ++i) {
                        vertx.eventBus().send(ADDRESS, "test " + i);
                    }
                });
    }

    @Override
    public void start(final Promise<Void> startPromise) {
        vertx.eventBus().localConsumer(ADDRESS, this::handleMessage);
        startPromise.complete();
    }

    private void handleMessage(Message<String> message) {
        if (handleMessageInExecution) {
            // this should never happen, since each thread that sets this to true, will also set it to
            // false on exit.
            System.out.println(message.body() + " ERROR");
            return;
        }

        handleMessageInExecution = true; // this thread is now executing handleMessage
        System.out.println(message.body() + " START   " + Thread.currentThread());

        try {
            Thread.sleep(1); // block thread for a moment to simulate heavy load
        } catch (Exception e) {
            // ignore interruption
            e.printStackTrace();
        } finally {
            handleMessageInExecution = false; // we are done executing
            System.out.println(message.body() + " END     " + Thread.currentThread());
        }
    }
}

我们看到了这个输出，这是预期的（每条消息由一个线程处理，它从头到尾运行，没有并发，最多同时有 2 条消息，因为我们有 2 个实例）：

test 1 START   Thread[vert.x-worker-thread-2,5,main]
test 0 START   Thread[vert.x-worker-thread-3,5,main]
test 0 END     Thread[vert.x-worker-thread-3,5,main]
test 1 END     Thread[vert.x-worker-thread-2,5,main]
test 2 START   Thread[vert.x-worker-thread-3,5,main]
test 3 START   Thread[vert.x-worker-thread-2,5,main]
test 3 END     Thread[vert.x-worker-thread-2,5,main]
test 2 END     Thread[vert.x-worker-thread-3,5,main]
test 5 START   Thread[vert.x-worker-thread-2,5,main]
test 4 START   Thread[vert.x-worker-thread-3,5,main]
test 4 END     Thread[vert.x-worker-thread-3,5,main]
test 6 START   Thread[vert.x-worker-thread-3,5,main]
test 5 END     Thread[vert.x-worker-thread-2,5,main]
test 7 START   Thread[vert.x-worker-thread-2,5,main]
test 6 END     Thread[vert.x-worker-thread-3,5,main]
test 8 START   Thread[vert.x-worker-thread-3,5,main]
test 7 END     Thread[vert.x-worker-thread-2,5,main]
test 9 START   Thread[vert.x-worker-thread-2,5,main]
test 8 END     Thread[vert.x-worker-thread-3,5,main]
test 10 START   Thread[vert.x-worker-thread-3,5,main]
test 9 END     Thread[vert.x-worker-thread-2,5,main]
test 11 START   Thread[vert.x-worker-thread-2,5,main]
test 10 END     Thread[vert.x-worker-thread-3,5,main]
...