把数百万的文件从蒙哥搬到卡夫卡

寻找最好的方法。

1.使用Mongo反应流

 Flux.from(collection.find()).doOnError(e -> {
             e.printStackTrace();
         }).doOnComplete(() -> {
             System.out.println("Finished ");
         }).subscribe(doc -> {
             // Code to insert into Kafka
         });

2. 使用Akka流

另外,我是否需要在subscribe方法中执行多线程?