如何使用Vert.x将MongoDB查询结果流式传输到Kafka主题？

这个需求刚好能发挥Vert.x异步流式API的优势，我给你梳理一套完整的实现思路，直接就能落地：

1. 用Vert.x MongoDB Client获取流式查询结果 #

首先你得用Vert.x官方的MongoDB客户端（不是原生Java同步驱动），它支持将查询结果以ReadStream<JsonObject>的形式返回，完美适配Vert.x的流生态。

步骤说明： #

• 先通过Vert.x的配置初始化MongoClient，可以用连接字符串或者配置项
• 构建你的查询条件（比如用Filters类的静态方法）
• 调用find()方法后，记得调用toStream()把异步查询转换成流式对象，这样就能逐行处理文档了

代码片段： #

// 初始化MongoClient
MongoClient mongoClient = MongoClient.createShared(vertx, new JsonObject()
  .put("connection_string", "mongodb://localhost:27017")
  .put("db_name", "your_database"));

// 构建查询条件，比如筛选status为active的文档
Bson query = Filters.eq("status", "active");

// 获取流式查询结果
ReadStream<JsonObject> mongoDocStream = mongoClient.find("your_collection", query).toStream();

2. 对接MongoDB流与KafkaWriteStream #

Vert.x的ReadStream和WriteStream天生就能对接，有两种方式：

方式一：用Pipe自动传输（最简单） #

Vert.x的Pipe类可以自动把ReadStream的数据转发到WriteStream，还会处理背压（backpressure），避免内存溢出：

// 假设你已经初始化好KafkaWriteStream<String, String> kafkaWriteStream
mongoDocStream.pipeTo(kafkaWriteStream, ar -> {
  if (ar.succeeded()) {
    System.out.println("所有文档已成功发送到Kafka");
  } else {
    System.err.println("流式传输失败：" + ar.cause().getMessage());
  }
});

方式二：手动处理每条记录（更灵活） #

如果需要在发送前修改文档（比如转换格式、添加字段），可以手动监听data事件：

mongoDocStream.handler(doc -> {
  // 将MongoDB的JsonObject转换成Kafka需要的消息格式，比如转成JSON字符串
  String messageValue = doc.encode();
  // 发送到Kafka，指定key（可选）
  kafkaWriteStream.write(new KafkaRecord<>("your_topic", "doc-" + doc.getString("_id"), messageValue));
});

// 处理流结束事件
mongoDocStream.endHandler(v -> {
  System.out.println("MongoDB文档流已读取完毕");
  // 可以在这里关闭Kafka流或者做其他清理
});

// 处理异常
mongoDocStream.exceptionHandler(err -> {
  System.err.println("MongoDB流读取出错：" + err.getMessage());
});

3. 完整的Verticle示例 #

把上面的逻辑整合到你的Verticle里，记得在Verticle启动/停止时管理资源：

public class MongoToKafkaVerticle extends AbstractVerticle {
  private MongoClient mongoClient;
  private KafkaWriteStream<String, String> kafkaWriteStream;

  @Override
  public void start(Promise<Void> startPromise) {
    // 初始化MongoClient
    mongoClient = MongoClient.createShared(vertx, new JsonObject()
      .put("connection_string", "mongodb://localhost:27017")
      .put("db_name", "test_db"));

    // 初始化KafkaWriteStream
    kafkaWriteStream = KafkaWriteStream.create(vertx, new KafkaClientOptions()
      .addBootstrapServer("localhost:9092"));

    // 构建查询并启动流式传输
    Bson query = Filters.gte("created_at", LocalDateTime.now().minusDays(7).atZone(ZoneId.systemDefault()).toInstant());
    ReadStream<JsonObject> mongoStream = mongoClient.find("orders", query).toStream();

    // 对接两个流
    mongoStream.pipeTo(kafkaWriteStream, ar -> {
      if (ar.succeeded()) {
        startPromise.complete();
        System.out.println("MongoDB到Kafka的流式传输已启动");
      } else {
        startPromise.fail(ar.cause());
      }
    });

    // 处理Kafka流的异常
    kafkaWriteStream.exceptionHandler(err -> {
      System.err.println("Kafka写入出错：" + err.getMessage());
    });
  }

  @Override
  public void stop() {
    // 关闭资源，避免泄漏
    if (mongoClient != null) mongoClient.close();
    if (kafkaWriteStream != null) kafkaWriteStream.close();
  }
}

4. 关键注意事项 #

• 索引优化：确保你的MongoDB查询条件有对应的索引，否则流式查询会很慢，甚至导致数据库压力过大
• 背压处理：Vert.x的流默认支持背压，但如果Kafka写入速度跟不上MongoDB读取速度，Pipe会自动暂停读取，不用担心内存爆炸
• 错误重试：如果Kafka写入失败，可以结合Vert.x的Retry工具类实现重试逻辑，比如：

  Retry.retry(5, 1000, id -> {
    kafkaWriteStream.write(record, ar -> {
      if (ar.failed()) id.retry(ar.cause());
      else id.complete();
    });
  });
  

• 资源清理：一定要在Verticle停止时关闭MongoClient和KafkaWriteStream，否则会留下连接泄漏

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