RxJava：为Completables列表指定最大并发数

嘿，这个场景我太熟悉了——之前做批量文件下载+API补全的时候也遇到过同样的需求：既要保持原来mergeDelayError那种“所有任务完成再通知，出错不中断其他任务”的特性，又要限制并发数防止把服务器打崩对吧？

直接用Completable.mergeDelayError确实没法控制并发，因为它会一次性订阅所有任务。这里给你两个最实用的解决方案，都是基于RxJava的原生操作符，不用额外依赖：

方法一：用Flowable+flatMapCompletable实现并发控制（推荐） #

这是最直接的方式，把你的Completable列表转换成Flowable，然后用flatMapCompletable指定最大并发数，同时保留延迟错误的特性：

// 假设你已经有了待执行的Completable任务列表
List<Completable> downloadAndFetchTasks = getYourCompletableList();

// 控制最大并发数为3，同时保持mergeDelayError的错误处理逻辑
Flowable.fromIterable(downloadAndFetchTasks)
    .flatMapCompletable(
        task -> task,
        true, // delayErrors=true：所有任务执行完再抛出错误，和mergeDelayError一致
        3     // maxConcurrency=3：同时最多执行3个任务
    )
    .subscribe(
        () -> System.out.println("所有下载/补全任务都完成啦！"),
        throwable -> System.err.println("任务执行出错（已延迟到所有任务完成后）：" + throwable.getMessage())
    );

参数说明： #

• delayErrors=true：和mergeDelayError的行为完全一致——哪怕某个任务失败，其他任务会继续执行，直到所有任务完成后才会抛出错误（如果有多个错误，会包装成CompositeException）。
• maxConcurrency=3：这里就是你要设置的最大并发数，按需调整就行。

方法二：单个任务出错时单独处理（不中断整体流程） #

如果不想最后才抛出汇总错误，而是希望单个任务失败时立刻处理（比如记录日志），同时不影响其他任务，只需要在flatMapCompletable里给单个任务加上onErrorResumeNext：

Flowable.fromIterable(downloadAndFetchTasks)
    .flatMapCompletable(
        task -> task.onErrorResumeNext(error -> {
            // 单个任务失败时的自定义处理：比如打印日志、记录失败任务
            System.err.println("任务执行失败：" + error.getMessage());
            // 返回Completable.complete()让整体流程继续
            return Completable.complete();
        }),
        true,
        3
    )
    .subscribe(
        () -> System.out.println("所有任务都处理完毕（包括失败的任务）"),
        throwable -> System.err.println("出现未处理的异常：" + throwable.getMessage())
    );

为什么不用Completable原生操作？ #

因为RxJava的Completable类本身没有提供带并发控制的merge操作，所以转成Flowable是最优雅的做法——Flowable的flatMap系列操作符天生支持并发数控制，而且能完美兼容Completable的错误处理逻辑。

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