VertX HTTP请求并行处理

Question

我正在使用Java VertX框架，并尝试使用VertX WebClient和一个简单的HTTP请求加载多个JSON对象。我想并行地做这件事，这样可以加快进程。

我有一个endpoint对象：

import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.codec.BodyCodec;

public final class Endpoint {

    private final String name;
    private final String url;

    public Endpoint (String name, String url) {
        this.name = name;
        this.url = url;
    }

    public String getName() {
        return name;
    }

    public String getUrl() {
        return url;
    }

    public JsonObject loadJsonObject() {
        WebClient client = WebClient.create(Vertx.vertx());
        client.getAbs(this.getUrl()).as(BodyCodec.jsonObject()).send(handler -> {
            // what to do
        });
        return null;
    }

}

在另一个类中，我有以下应该并行处理的函数（源代码）：

public static void useCompletableFutureWithExecutor(List<Endpoint> tasks) {
    long start = System.nanoTime();
    ExecutorService executor = Executors.newFixedThreadPool(Math.min(tasks.size(), 10));
    List<CompletableFuture<JsonObject>> futures =
        tasks.stream()
             .map(t -> CompletableFuture.supplyAsync(() -> t.loadJsonObject(), executor))
             .collect(Collectors.toList());

    List<JsonObject> result =
        futures.stream()
               .map(CompletableFuture::join)
               .collect(Collectors.toList());
    long duration = (System.nanoTime() - start) / 1_000_000;
    System.out.printf("Processed %d tasks in %d millis\n", tasks.size(), duration);
    System.out.println(result);
    executor.shutdown();
}

我不知道如何继续下去。VertX WebClient强制我使用异步处理程序，这意味着我不能直接返回JsonObject。

Answer 1

您的问题在于方法签名。让我们从public Future开始，而不是public JsonObject loadJsonObject（）{

这意味着不是返回null，而是返回null

剩下的是在数据到达时放置数据：

future.complete(result);

最终结果如下：

public Future<JsonObject> loadJsonObject() {
        WebClient client = WebClient.create(Vertx.vertx());
        Future<JsonObject> future = Future.future();
        client.getAbs(this.getUrl()).as(BodyCodec.jsonObject()).send(handler -> {
            // Do something with JSON and put in result
            future.complete(result);

            // Remember to future.fail() in case something went wrong
        });
        return future;
    }

顺便说一句，您还可以使用CompositeFuture。all（）等待您的所有未来。

http://vertx.io/docs/vertx-core/java/#_async_coordination