为什么是垂直。从多个线程并发调用x worker verticle？

Question

我的vertx（4.0.2）应用程序是用Java（11）编写的，它使用了一些数据量大的垂直项，这些垂直项会导致延迟峰值，因为eventloop会被它们阻塞一段时间。出于这个原因，我想将这些垂直链接部署为工作垂直链接，这样eventloop和其他垂直链接就不再被阻塞。

不幸的是，我的应用程序现在崩溃了，因为顶点内的事件处理是由多个线程并发执行的；（

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

Vert从不并发执行Worker verticle实例。x由多个线程执行，但可以在不同的时间由不同的线程执行。

我用一个简单的例子再现了这个问题：

@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。如果我以非工作方式部署垂直体，这将按预期工作。

我做错了什么？

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]
...