Vertx.io中串联可变数量异步回调并收集结果的正确方式

嘿，作为Vert.x新手碰到这种并行异步任务收集结果的问题太正常了！我来给你说说正确的实现方式，Vert.x本身就提供了专门的工具来处理这种场景，不用自己手动去计数等待～

基础实现方式（兼容所有Vert.x版本） #

我们可以用CompositeFuture来等待所有异步操作完成，然后统一收集结果：

import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.AsyncResult;
import java.util.ArrayList;
import java.util.List;

public void doSomething(String[] params, Handler<AsyncResult<List<Stuff>>> h) {
    // 第一步：创建列表存放每个异步操作的Future实例
    List<Future<Stuff>> futures = new ArrayList<>();
    
    for (String param : params) {
        // 把每个doSomethingElse调用包装成Future
        Future<Stuff> future = Future.future(promise -> {
            doSomethingElse(param, ar -> {
                if (ar.succeeded()) {
                    promise.complete(ar.result());
                } else {
                    promise.fail(ar.cause());
                }
            });
        });
        futures.add(future);
    }
    
    // 第二步：等待所有Future完成
    CompositeFuture.all(futures)
        .onComplete(ar -> {
            if (ar.succeeded()) {
                // 第三步：从CompositeFuture中提取所有结果并组装成List
                List<Stuff> resultList = new ArrayList<>();
                for (int i = 0; i < futures.size(); i++) {
                    resultList.add(ar.result().resultAt(i));
                }
                // 通过传入的Handler返回成功结果
                h.handle(Future.succeededFuture(resultList));
            } else {
                // 只要有一个操作失败，就传递失败原因
                h.handle(Future.failedFuture(ar.cause()));
            }
        });
}

Vert.x 4+ 简洁写法 #

如果你用的是Vert.x 4及以上版本，可以用Futures.all()方法，代码会更简洁，它会直接返回一个包含结果列表的Future：

import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.AsyncResult;
import io.vertx.core.Futures;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public void doSomething(String[] params, Handler<AsyncResult<List<Stuff>>> h) {
    // 将参数数组转换成Future列表
    List<Future<Stuff>> futures = Arrays.stream(params)
        .map(param -> Future.future(promise -> doSomethingElse(param, promise)))
        .collect(Collectors.toList());
    
    // 等待所有操作完成，直接获取结果列表
    Futures.all(futures)
        .onComplete(ar -> {
            if (ar.succeeded()) {
                h.handle(Future.succeededFuture(ar.result()));
            } else {
                h.handle(Future.failedFuture(ar.cause()));
            }
        });
}

额外提示 #

• 如果你的场景允许部分操作失败（不需要一个失败就终止全部），可以把CompositeFuture.all()换成CompositeFuture.join()，或者Futures.join()，这样即使有失败，也能拿到所有成功的结果，同时可以检查哪些操作失败了。
• 一定要确保doSomethingElse的Handler正确处理成功和失败的情况，否则会有Future一直处于等待状态，导致整个任务永远无法完成。

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