RxJava中与VertX CompositeFuture的等价实现是什么？附VertX示例

当然有啦！RxJava里有好几种实现可以对应VertX的CompositeFuture这类功能，下面给你拆解最常用的几种方案：

1. Single.zip()（最贴合CompositeFuture.all()的场景） #

如果你的异步操作都是返回Single（和VertX的Future语义一致，代表单个异步结果），那么Single.zip()就是最直接的等价实现——它会等待所有Single都成功完成，然后把所有结果合并在一起返回；只要有任意一个操作失败，就会立刻触发错误回调，这和CompositeFuture.all()的行为完全匹配。

举个和你VertX示例对应的代码：

// 先把VertX的异步操作包装成Single
Single<HttpServer> httpServerSingle = Single.create(emitter -> {
    httpServer.listen(res -> {
        if (res.succeeded()) {
            emitter.onSuccess(res.result());
        } else {
            emitter.onError(res.cause());
        }
    });
});

Single<NetServer> netServerSingle = Single.create(emitter -> {
    netServer.listen(res -> {
        if (res.succeeded()) {
            emitter.onSuccess(res.result());
        } else {
            emitter.onError(res.cause());
        }
    });
});

// 合并两个Single，对应CompositeFuture.all()的逻辑
Single.zip(httpServerSingle, netServerSingle, (httpServer, netServer) -> {
    // 这里可以把两个结果封装成自定义对象，或者直接处理
    return new ServerPair(httpServer, netServer);
}).subscribe(
    serverPair -> {
        // 所有服务器启动成功，处理逻辑
    },
    throwable -> {
        // 至少一个服务器启动失败，处理错误
    }
);

2. Completable.mergeArray()（无返回值的场景） #

如果你的异步操作不需要返回结果，只是要等待所有操作完成（类似VertX中不带结果的Future），可以用Completable.mergeArray()。它会等待所有Completable执行完成，只要有一个失败就会触发错误回调，完美对应这类场景。

3. Observable.merge()（注意语义差异） #

Observable.merge()也可以合并多个异步操作，但它会按操作完成的顺序逐个发射结果，而不是等待所有操作完成后统一返回。如果你的场景只是要确保所有操作完成，但不需要合并结果，它也能工作，但语义上和CompositeFuture.all()有区别——zip更贴合“等待所有完成后统一处理”的需求。

另外，如果你用的是RxJava 3版本，Flowable.zip()等对应方法的逻辑也是完全一致的，核心都是等待所有异步任务完成，统一处理成功或失败的结果。

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