如何在 RxJava2 中批量处理链中的 Flowable 输出答案

【问题标题】：How to batch process a Flowable output in a chain in RxJava2如何在 RxJava2 中批量处理链中的 Flowable 输出
【发布时间】：2023-03-07 18:11:01
【问题描述】：

如何在 RxJava2 中链接要批量处理的进程。下面的流程图是我想要实现的。

Flowable#1          Flowable#2 (process every 10)
==============     ================================
callServer(p1) ->
      :        ->  saveToDatabase(List<r1 to r10>)
callServer(p20)->  saveToDatabase(List<r11 to r20>)
callServer(p21)->      :
      :                :
callServer(p35)->  saveToDatabase(List<r31 to r35>) //the remainder

目前，我要做的是等待所有结果返回，然后再保存到数据库中。

Flowable.fromIterable(paramList)
    .map(p -> callServer(p)) 
    //wait for the return a map of ALL the results r  
    //how to chain it such that saveToDatabase process after 'n' results 
    .toList()  
    .flatmap(listOfR -> saveToDatabase(listOfR);

如何使 saveToDatabase 在每个“n”个结果之后被调用，而不是等待所有结果完成？

【问题讨论】：

标签： rx-java rx-java2

【解决方案1】：

使用buffer() 操作符和n 作为参数，缓冲区将从源Observable 收集n 个项目，并发出包含n 个项目的列表。
所以你可以一次处理每n个项目，并将它们保存到数据库中

【讨论】：

这是正确答案。此外，如果您需要每 n 秒处理一次，则可以使用窗口运算符之一。