响应式并发批处理

假设DataProcessor接口定义了方法batchProcess能够对一批数据进行处理，一批处理500个数据。现在我们需要对一个响应式数据流 Flux dataItems 调用 batchProcess() 进行处理。

public interface DataProcessor {
  Mono<String> batchProcess(List<DataItem> dataItems);
  ... ...
}


DataProcessor dataProcessor = ...;

int batchSize = 500;

Flux<DataItem> dataItems = ...

下面分别串行和并行的方式展示一下Reactor API的使用。

1）攒够 batchSize 个数据后进行处理。

这里关键是buffer方法的使用。

Mono<List<String>> result = dataItems.buffer(batchSize)
    .flatMap(dataProcessor::batchProcess)
    .collectList();

2）以并行的方式，把流分成10股，每股攒够 batchSize 个数据后进行处理。

Mono<List<String>> result = dataItems
    .parallel(10)
    .runOn(Schedulers.fromExecutor(Executors.newFixedThreadPool(10)))
    .groups()
    .flatMap(g -> g.buffer(batchSize).flatMap(dataProcessor::batchProcess))
    .collectList();

这里runOn接收的参数可以是Schedulers不同策略的实现，具有不同适用范围，比如适合计算密集型的ParallelScheduler、单线程的SingleScheduler。这里使用的是Executors FixedThreadPool。

可以想象如果我们自己实现这样一个处理逻辑的复杂度，而通过reactor api，仅仅几行代码就完成了这么复杂高效的处理。

3）使用 reactive mongo driver需要的线程。

Spring默认到monog的链接池最大为100，但是实际上在使用reactive方式访问时使用20～10个左右的线程就足够了。因此对mongog的连接串最好明确使用适合自己情况的连接数以避免连接浪费或不够。 测试了一个70万条、大概250M数据的批量插入，发现无论使用串行还是并行，数据库插入时间都差不多（36s～26s）。而连接池最大连接设为200、100、50、20、10对数据库插入的性能也没有太大影响，200个线程时反而有一点下降。这个情况从mongo响应式驱动的角度去解释是完全可以理解的，如果使用传统驱动，恐怕所需的线程就不是这个量级的了。