如何重构 vertx 中的异步调用链以避免回调地狱

Question

我有以下代码，其中有几个相互依赖的异步调用（例如，调用可以是 api REST），最后处理所有结果。这是我的示例代码：

private void foo1(String uuid, Handler<AsyncResult<JsonObject>> aHandler) {
    //call json api, for example
    JsonObject foo1 = new JsonObject();
    foo1.put("uuid", "foo1");
    aHandler.handle(Future.succeededFuture(foo1));
 }

private void foo2(String uuid, Handler<AsyncResult<JsonObject>> aHandler) {

    //call json api, for example
    JsonObject foo2 = new JsonObject();
    foo2.put("uuid", "foo2");
    aHandler.handle(Future.succeededFuture(foo2));
 }

private void foo3(String uuid, Handler<AsyncResult<JsonObject>> aHandler) {

    //call json api, for example
    JsonObject foo3 = new JsonObject();
    foo3.put("uuid", "foo3");
    aHandler.handle(Future.succeededFuture(foo3));
 }

private void doSomething(JsonObject result1, JsonObject result2, JsonObject result3, Handler<AsyncResult<JsonObject>> aHandler) {
    JsonObject finalResult =new JsonObject();
    aHandler.handle(Future.succeededFuture(finalResult));
}

private void processToRefactor (String uuid, Handler<AsyncResult<JsonObject>> aHandler) {

    foo1(uuid, ar -> {
        if (ar.succeeded()) {
            JsonObject foo1 = ar.result();
            foo2(foo1.getString("uuid"), ar2 ->{
                if (ar2.succeeded()) {
                    JsonObject foo2 = ar2.result();
                    foo3(foo2.getString("uuid"), ar3 -> {
                        if (ar3.succeeded()) {
                            JsonObject foo3 = ar3.result();
                            doSomething(foo1, foo2, foo3, aHandler);
                        } else {
                            ar3.cause().printStackTrace();
                        }
                    });
                } else {
                    ar2.cause().printStackTrace();
                }
            });
        } else {
            ar.cause().printStackTrace();
        }

    });
}

在前面的代码中，如果所有调用都成功，我可以在 "doSomething" 方法中使用所有结果。我尝试使用以下 link https://streamdata.io/blog/vert-x-and-the-async-calls-chain/

中的简单 "HelloWord" 示例重构此代码

这是我的结果：

    private void process(String uuid, Handler<AsyncResult<JsonObject>> aHandler) {

        Future<JsonObject> future = Future.future();
        future.setHandler(aHandler);

        Future<JsonObject> futureFoo1 = Future.future();

        foo1(uuid, futureFoo1);

        futureFoo1.compose(resultFoo1 -> {
            Future<JsonObject> futureFoo2 = Future.future();
            foo2(resultFoo1.getString("uuid"), futureFoo2);

            return futureFoo2; 
        }).compose(resultFoo2 ->{
            Future<JsonObject> futureFoo3 = Future.future();
            foo3(resultFoo2.getString("uuid"), futureFoo3);

            return futureFoo3;

        }).compose(resultFoo3 -> {

            // How to get result1, result2 and result3?
//            doSomething(resultFoo1, resultFoo2, resultFoo3, aHandler);

        }, future);
    }

新代码更简洁明了，但是在使用 compose 时，在调用函数时 "doSomething" 我没有所有可用的调用结果。 如何获取链末端的所有结果？

另一方面，如果其中一个api调用方法returns一个数组怎么办？也就是说，对于数组的每个元素, 应用了一系列功能，独立于一些有结果而另一些没有的事实。例如：

private void foo1Array(String uuid, Handler<AsyncResult<JsonArray>> aHandler) {

    //call json api that return array, for example
    JsonArray result = new JsonArray();
    JsonObject foo1 = new JsonObject();
    foo1.put("uuid", "foo1");

    JsonObject foo2 = new JsonObject();
    foo1.put("uuid", "foo2");

    JsonObject foo3 = new JsonObject();
    foo1.put("uuid", "foo3");

    result.add(foo1);
    result.add(foo2);
    result.add(foo3);

    aHandler.handle(Future.succeededFuture(result));
 }   
private void processArray(String uuid, Handler<AsyncResult<JsonObject>> aHandler) {

    Future<JsonObject> future = Future.future();
    future.setHandler(aHandler);

    Future<JsonArray> futureFoo1 = Future.future();

    foo1Array(uuid, futureFoo1);

    futureFoo1.compose(resultArray -> {
        List<Future> futures = new ArrayList<Future>();
        for (int i = 0; i < resultArray.size(); i ++) {
            JsonObject resultFoo1 = resultArray.getJsonObject(i);

            Future<JsonObject> futureFoo2 = Future.future();
            foo2(resultFoo1.getString("uuid"), aHandler);
            futures.add(futureFoo2);
        }

        CompositeFuture.any(futures).setHandler(ar -> {
            //What to do here?
        });

    }, future);
}

如何使用 foo1Array 的结果调用函数 foo2、foo3、...，然后在 doSomething 中使用它？

Answer 1

如果你想要更简单且没有回调地狱的代码，那么切换到 rx，你上面的代码将变成这样的：

private Single<JsonObject> foo1(String uuid) {
    JsonObject foo1 = new JsonObject();
    foo1.put("uuid", "foo1");
    return Single.just(foo1);
 }

private Single<JsonObject> foo2(String uuid) {
    JsonObject foo2 = new JsonObject();
    foo2.put("uuid", "foo2");
    return Single.just(foo2);
 }

private Single<JsonObject> foo3(String uuid) {
    JsonObject foo3 = new JsonObject();
    foo3.put("uuid", "foo3");
    return Single.just(foo3);
 }

private Single<JsonObject> doSomething(String uuid) {
    JsonObject finalResult =new JsonObject();
    return Single.zip(foo1(uuid), foo2(uuid), foo3(uuid))
    .map(results -> {
        // Map zip results to finalResult
        return finalResult;
    });
}

private void processToRefactor (String uuid, Handler<AsyncResult<JsonObject>> aHandler) {
    doSomething(uuid).subscribe();
}

Answer 2

你最初的方法其实还不错。

为了更好地改进代码 "composability"，您应该将每个 fooX 方法的处理程序输入参数更改为扩展 Handler<AsyncResult<JsonObject>>（例如 Future）和 returns结果是相同的处理程序，因此它在 `Future.compose 中变得更好用，因为传入的处理程序可以用作每个组合的 return 值：

 private <T extends Handler<AsyncResult<JsonObject>>> T foo1(String uuid, T aHandler) {
    JsonObject foo1 = new JsonObject().put("uuid", "foo1");
    aHandler.handle(Future.succeededFuture(foo1));
    return aHandler; //<-- return the handler here
}

其次，为了在最后阶段访问所有三个结果，你必须在链外声明三个期货。现在您可以使用每个 foo 方法的输出作为每个 compose 的结果很好地链接期货。

Future<JsonObject> futureFoo1 = Future.future();
Future<JsonObject> futureFoo2 = Future.future();
Future<JsonObject> futureFoo3 = Future.future();


foo1(uuid, futureFoo1).compose(resultFoo1 -> foo2(resultFoo1.getString("uuid"), futureFoo2))
                      .compose(resultFoo2 -> foo3(resultFoo2.getString("uuid"), futureFoo3))
                      .compose(resultFoo3 -> doSomething(futureFoo1.result(), //access results from 1st call
                                                         futureFoo2.result(), //access results from 2nd call 
                                                         resultFoo3,
                                                         Future.<JsonObject>future().setHandler(aHandler))); //pass the final result to the original handler

如果您不能接受这种方法的 "impurity"（在链外定义期货并在函数内修改它们），您必须为每个方法传递原始输入值（=输出之前调用的）以及结果，但我怀疑这会使代码更具可读性。

为了在一个 compose 方法中改变类型，你的 fooX 方法必须使转换，不是 return 原始处理程序，而是具有不同类型的新 Future

private Future<JsonArray> foo2(String uuid, Handler<AsyncResult<JsonObject>> aHandler) {
    JsonObject foo2 = new JsonObject();
    foo2.put("uuid", "foo2" + uuid);
    aHandler.handle(Future.succeededFuture(foo2));
    JsonArray arr = new JsonArray().add("123").add("456").add("789");
    return Future.succeededFuture(arr);
}

如何重构 vertx 中的异步调用链以避免回调地狱

How to refactor chain of asynchronous calls in vertx to avoid the callback hell

java

asynchronous

callback

vert.x