vert.x mongo客户端模块允许 ReadStream 来自mongo查询。
当你拥有它的时候 KafkaWriteStream ，您可以开始从mongo读取数据并将其发送给kafka。
不过，要小心背压：如果mongo加载数据太快，不要让kafka客户端崩溃。你的算法应该是这样的：

readstream.handler(data -> {
  transformedData = transform(data); // your own transformations
  writestream.write(transformedData);
    if (writestream.writeQueueFull()) {
      readstream.pause();
      writestream.drainHandler(done -> {
        readstream.resume();
      });
    }
});