为什么 Vertx Event Bus 在高负载下阻塞？

Question

我正在尝试通过Vertx事件总线发送大量消息，如下所示（与Hazelcast集群），而不阻塞：

EventBus eb = vertx.eventBus();

for (int i = 0; i < 100; i++) {
  vertx.setPeriodic(1, num -> {
    eb.send("clusteredEndpoint", "ping");
  });
}

当计时器的数量较小时，它可以正常工作，但在大约100个计时器时，我得到了这个错误。

我想知道如何在不阻塞的情况下扩展到100K事件/秒（作为参考，我编写了一个Vertex WebSocket测试，可能会超过这个数字）。

如果不可能，我想知道是什么阻塞了——看起来是这个类中的某个东西:https://github . com/eclipse-vertx/vert . x/blob/master/src/main/Java/io/vertx/core/event bus/impl/clustered/serializer . Java

对于参考-此代码不会阻塞-即使使用1000个定时器：

HttpClient client = vertx.createHttpClient();
client.webSocket(8080, "localhost", "/", res -> {
  for (int i = 0; i < 1000; i++) {
    vertx.setPeriodic(1, num -> {
      res.result().writeTextMessage("ping");
    });
  }
});
});

2020 年 12 月 15 日 10：54：38 AM io.vertx.core.impl.BlockedThreadChecker 警告： Thread Thread[vert.x-eventloop-thread-1，5，main] 已被阻止 36794 毫秒，时间限制为 2000 毫秒 io.vertx.core.Vertx.core.impl.future.Future.FutureImpl.addListener（FutureImpl.java：140） at io.vertx.core.impl.future.PromiseImpl.addListener（PromiseImpl.java：23） at io.vertx.core.impl.future.FutureImpl.onComplete（FutureImpl.java：133） at io.vertx.core.impl.future.PromiseImpl.onComplete（PromiseImpl.java：23） at io.vertx.core.spi.cluster.impl.selector.Selectors.withSelector（Selectors.java：48） at io.vertx.core.spi.cluster.impl.DefaultNodeSelector.selectForSend（DefaultNodeSelector.java：42） at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus$$Lambda$1065/195695453.accept（Unknown Source） at io.vertx.core.eventbus.impl.clustered.Serializer$SerializerQueue$SerializedTask.process（Serializer.java：147） atio.vertx.core.eventbus.impl.clustered.Serializer$SerializerQueue.checkPending（Serializer.java：94） at io.vertx.core.eventbus.impl.clustered.Serializer$SerializerQueue.add（Serializer.java：114） at io.vertx.core.eventbus.impl.clustered.Serializer.queue（Serializer.java：65） at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendOrPub（ClusteredEventBus.java：172） at io.vertx.core.eventbus.impl.OutboundDeliveryContext.next（OutboundDeliveryContext.java：127） at io.vertx.core.eventbus.impl.EventBusImpl.sendOrPubInternal（EventBusImpl.java：394） at io.vertx.core.eventbus.impl.eventBusImpl.sendOrPubInternal（EventBusImpl.java：400） at io.vertx.core.eventbus.impl.EventBusImpl.send（EventBusImpl.java：103） at io.vertx.core.eventbus.impl.EventBusImpl.send（EventBusImpl.java：97） at io.vertx.example.EBtestClient.html" target="_blank">lambda$start$0（EBtestClient.java：22） at io.vertx.example.EBtestClient$$Lambda$1056/1487417027.handle（Unknown Source） at io.vertx.core.impl.VertxImpl$InternalTimerHandler.handle（VertxImpl.java：939） at io.vertx.core.impl.VertxImpl$InternalTimerHandler.handle（VertxImpl.java：910） at io.vertx.core.impl.EventLoopContext.emit（EventLoopContext.java：52） at io.vertx.core.impl.ContextImpl.emit（ContextImpl.java：294） at io.vertx.core.impl.EventLoopContext.emit（EventLoopContext.java：24） at io.vertx.core.impl.AbstractContext.emit（AbstractContext.java：49） atio.vertx.core.impl.EventLoopContext.emit（EventLoopContext.java：24） at io.vertx.core.impl.VertxImpl$InternalTimerHandler.run（VertxImpl.java：933） at io.netty.util.concurrent.PromiseTask.runTask（PromiseTask.java：98） at io.netty.util.concurrent.ScheduledFutureTask.run（ScheduledFutureTask.java：176） at io.netty.util.concurrent.AbstractEventExecutor.safeExecute（AbstractEventExecutor.java：164） at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks（SingleThreadEventExecutor.java：472） at io.netty.channel.nio.nioEventLoop.run（NioEventLoop.java：500） at io.netty.util.concurrent.SingleThreadEventExecutor$4.run（SingleThreadEventExecutor.java：989） at io.netty.util.internal.ThreadExecutorMap$2.run（ThreadExcutorMap.java：74） at io.netty.util.concurrent.FastThreadLocalRunnable.run（FastThreadLocalRunnable.java：30） at java.lang.Thread.run（Thread.java：748）

Answer 1

以下是我在进一步调查后的分析：

当使用Vertx事件总线进行远程通信时，一旦消费者感到不知所措，它就会停止响应。这会导致生产者阻塞，我已经捕获了3条不同的阻塞消息（见下文）。阻塞警告后有以下警告：

警告：没有来自服务器2d1fb2ce-940f-4b60-bf60-39847f31bcaf的pong-将认为它已死

我问题的答案是，它阻止“为什么”并不重要，因为它是死的（因为它达到了一定的限制）。

我很惊讶Vert。x不会更优雅地处理这个问题-比如可能会抛出一个异常。

阻塞错误#1

线程在io.vertx.core.impl.future.FutureImpl.addListener（FutureImpl.java:140）和io.Vertex.Cor.impl.future.PromiseImpl.AddListen（PromiseImpl.java:23）处被阻塞，在io.vertx.core.impl.future.FutureMimpl.onComplete（FutureMImpl.java:133）和io.vertx.core.imp.future.PromiseMimpl.onComplete（PromiseMImpl.java:23）处和io.vertex.core.spi.cluster.impl.selector.Selectors处被阻塞。Selectors.java:48）在

阻塞错误#2

io.vertx.core.VertxException:线程阻塞在Java . nio . charset . charset Encoder .(charset Encoder . Java:198)在Java . nio . charset . charset Encoder .(charset Encoder . Java:233)在sun . nio . cs . utf _ 8＄Encoder .(UTF_8.java:558)在sun.nio.cs.UTF_8$Encoder。(UTF_8.java:554)在sun . nio . cs . utf _ 8 . new encoder(UTF _ 8 . Java:72)

阻塞错误#3

io.vertx.core.VertxException：线程io.vertx.core.eventbus.impl.clustered.ConnectionHolder.writeMessage（ConnectionHolder.java:93）io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendRemote（ClusteredEventBus.java:332）io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendToNode（ClusteredEventBus.java:283）

Answer 2

首先，您将在同一个线程上运行100个任务，因为Vert. x具有线程亲和性。如果您想避免这种情况，请在不同的顶点上运行它们。但是，我认为您没有100个CPU，所以会有很多争用。

将它们全部设置为每 1 毫秒执行一次意味着它们需要以某种方式在每个 10 微秒内完成，其中包括网络代码，因为您使用的是群集 EventBus。

所以，这是测试的编写方式，而不是Vert.x正在做的事情。

如果您真的想测试这种负载(我们这里说的是100K rps)，请将您的请求分散到多台机器上。

不过，我不确定Hazelcast是否能承受这种负载。

如果您想知道真正阻塞的是什么，我猜是这部分代码:

https://github.com/eclipse-vertx/vert.x/blob/master/src/main/java/io/vertx/core/spi/cluster/impl/DefaultNodeSelector.java#L43

由于我没有一个集群的Vert. x很容易设置，我无法确认我的假设是否正确。