Vert.x事件总线容量问题：文件推Kafka时Verticle1日志停止输出

从你的描述来看，问题的核心确实是事件总线消息堆积导致生产者（Verticle1）阻塞——因为你没有使用工作线程，Verticle1是在事件循环线程上运行的，当事件总线的消息队列被占满后，默认的消息发送行为会让事件循环陷入等待，最终导致Verticle1无法继续读取文件和输出日志，同时还要保证文件行的顺序不被打乱，这就需要在不破坏顺序的前提下实现流量控制。

下面是针对性的解决方案：

1. 先确认事件总线队列配置 #

Vert.x的事件总线默认有队列大小限制（默认值通常是10000），你可以先调整这个参数临时缓解，但这只是治标不治本：

// 创建Vertx实例时配置事件总线队列大小
Vertx vertx = Vertx.vertx(new VertxOptions()
    .setEventBusOptions(new EventBusOptions()
        .setQueueSize(50000)));

2. 实现基于顺序的背压机制 #

要保证消息顺序，不能用并行处理，所以必须让Verticle1的发送速度匹配Verticle2的处理速度，推荐两种实用方式：

方式一：基于异步回调的串行发送 #

Verticle1每发送一条消息，等待事件总线的回调确认后，再读取并发送下一行。这种方式严格保证顺序，且绝对不会让事件总线堆积：

// Verticle1 核心代码示例
vertx.fileSystem().open("target-file.txt", new OpenOptions(), res -> {
  if (res.succeeded()) {
    AsyncFile file = res.result();
    BufferedReader reader = new BufferedReader(new InputStreamReader(file.getInputStream()));

    // 定义递归发送方法，保证串行顺序
    Consumer<Void> sendNextLine = v -> {
      try {
        String line = reader.readLine();
        if (line == null) {
          // 文件读取完毕，关闭资源
          file.close();
          return;
        }
        // 执行你的过滤逻辑
        if (isLineValid(line)) {
          // 发送到事件总线，等待回调后再发下一条
          vertx.eventBus().send("kafka-publish-topic", line, ar -> {
            if (ar.succeeded()) {
              System.out.println("Processed line: " + line);
              sendNextLine.accept(null);
            } else {
              // 处理发送失败，比如重试或记录日志
              ar.cause().printStackTrace();
              sendNextLine.accept(null);
            }
          });
        } else {
          // 跳过不符合条件的行，继续下一个
          sendNextLine.accept(null);
        }
      } catch (IOException e) {
        e.printStackTrace();
      }
    };

    // 启动串行发送流程
    sendNextLine.accept(null);
  } else {
    res.cause().printStackTrace();
  }
});

方式二：基于流量控制的批量发送 #

如果串行发送的吞吐量达不到要求，可以维护一个待发送队列，同时让Verticle2在处理完一批消息后通知Verticle1继续发送，既保证顺序又能提升效率：

// Verticle1 核心逻辑
private final Queue<String> pendingLines = new LinkedList<>();
private boolean canSend = true;
private static final int BATCH_LIMIT = 100;

// 初始化时订阅流量控制信号
vertx.eventBus().consumer("flow-control-allow-send", msg -> {
  canSend = true;
  processPendingBatch();
});

// 读取文件行后加入待发送队列
private void enqueueLine(String line) {
  pendingLines.add(line);
  if (canSend && pendingLines.size() >= BATCH_LIMIT) {
    processPendingBatch();
  }
}

// 批量发送待处理的行
private void processPendingBatch() {
  canSend = false;
  int sentCount = 0;
  while (!pendingLines.isEmpty() && sentCount < BATCH_LIMIT) {
    String line = pendingLines.poll();
    vertx.eventBus().send("kafka-publish-topic", line);
    sentCount++;
  }
}

// Verticle2 核心逻辑
private final AtomicInteger processedCount = new AtomicInteger(0);

vertx.eventBus().consumer("kafka-publish-topic", msg -> {
  String line = msg.body().toString();
  // 异步发送到Kafka，避免阻塞事件循环
  kafkaProducer.write(new ProducerRecord<>("your-kafka-topic", line), ar -> {
    if (ar.succeeded()) {
      // 每处理100条，发送流量控制信号允许Verticle1继续发送
      if (processedCount.incrementAndGet() % BATCH_LIMIT == 0) {
        vertx.eventBus().send("flow-control-allow-send", "continue");
      }
      msg.reply("processed");
    } else {
      // 处理发送失败逻辑，比如重试或告警
      ar.cause().printStackTrace();
    }
  });
});

3. 优化Verticle2的Kafka发送效率 #

确保Kafka发送操作不会阻塞事件循环：

• 必须使用Vert.x Kafka客户端的异步write方法，绝对不要同步等待发送结果
• 合理配置Kafka生产者的批量参数（比如batch.size、linger.ms），在保证顺序的前提下提升发送效率；注意acks参数建议设为1或all，避免消息丢失同时保证顺序

4. 确保文件读取是异步的 #

一定要用Vert.x的异步文件IO，避免在事件循环中做同步读文件操作——同步IO会直接阻塞事件循环，这也是导致Verticle1停滞的常见坑：

// 正确的异步逐行读取方式
vertx.fileSystem().open("target-file.txt", new OpenOptions(), res -> {
  if (res.succeeded()) {
    AsyncFile file = res.result();
    file.setReadBufferSize(4096);
    // 读取文件内容并分割成行
    file.handler(buffer -> {
      String content = buffer.toString();
      String[] lines = content.split("\n");
      for (String line : lines) {
        // 调用你的行处理逻辑
        enqueueLine(line);
      }
    });
    // 文件读取完毕的回调
    file.endHandler(v -> {
      file.close();
      // 发送剩余的待处理行
      processPendingBatch();
    });
  }
});

总结 #

核心思路是在保证顺序的前提下实现生产者和消费者的速度匹配，通过背压机制避免事件总线堆积，同时确保所有操作都在异步非阻塞的模式下运行，这样就能彻底解决Verticle1停滞的问题。

内容的提问来源于stack exchange，提问作者Yogi