Vert.x 中 TCP 服务器的问题

Question

我的系统是同一个Node同一个JVM的高速流系统，我想用NetServer接收消息，用事件总线转发消息到同一个Vertx中的其他Verticle。

众所周知，在 Vert.x 中，NetServer 由事件循环线程调用。我遇到了一个问题。假设事件总线消费者很慢，而 NetServer 太快无法向事件总线发送数据，如果消费者缓冲区已满，事件总线将丢弃并丢失数据。所以我在想是否有什么方法可以在 NetServer 中进行流量控制？或者事件总线对 NetServer 有背压，以减慢发送到事件总线的速度。

我很欣赏你的想法和 cmets。

Answer 1

据我所知，Vert.x EventBus 不会为本地消费者丢弃消息。

虽然文档指出DEFAULT_ACCEPT_BACKLOG 设置为 1024，但实际上是 -1：

https://github.com/eclipse-vertx/vert.x/blob/1ab9884cb29adea75b378c3fa7359813f3de68f9/src/main/java/io/vertx/core/net/NetServerOptions.java#L45

而且我没有看到它被用于任何地方来限制 EventBus 的大小。

您还可以使用此测试来确保所有消息都实际到达：

public class BackpressureExample {

    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx(new VertxOptions().setEventBusOptions(new EventBusOptions().setAcceptBacklog(1)));

        vertx.deployVerticle(new SlowConsumer(), new DeploymentOptions().setWorker(true), h -> {
            for (int i = 0; i < 10_000; i++) {
                vertx.eventBus().send("address", UUID.randomUUID().toString());
            }
            System.out.println("Done producing");
        });
    }

    private static class SlowConsumer extends AbstractVerticle {

        @Override
        public void start() {
            AtomicInteger counter = new AtomicInteger();
            this.vertx.eventBus().consumer("address", h -> {
                System.out.println(counter.incrementAndGet());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

您可以在此处查看消息是如何传递给本地消费者的： https://github.com/eclipse-vertx/vert.x/blob/master/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java#L359