无法将可完成未来的结果输入响应对象

Question

我使用Java8的Java.util.concurrent的CompletableFuture接口调用了四个API。我想执行多个rest调用，组合结果并返回一个JSON。

public Map<String, Map<String, Object>> getAllValuesInParallel2(RequestObj requestObj) {

    Map<String, Map<String, Object>> response = new HashMap<>();
    ExecutorService executor = Executors.newFixedThreadPool(6);
    CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("mani");
        return aClient.transform(keys, requestObj);
    }, executor).thenApply(s -> {
        putIntoResponse(response, s);
        return true;
    });

    CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("gani");
        return bClient.transform(keys, requestObj);
    }, executor).thenApply(s -> {
        putIntoResponse(response, s);
        return true;
    });

    CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("priya");
        return cClient.transform(keys, requestObj);
    }, executor).thenApply(s -> {
        putIntoResponse(response, s);
        return true;
    });

    CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("ravi");
        return dClient.transform(keys, requestObj);
    }, executor).thenApply(s -> {
        putIntoResponse(response, s);
        return true;
    });

    return response;
}

private void putIntoResponse(Map<String, Map<String, Object>> response, List<Map<String, Object>> s) {
    if(s.size() > 0) {
        for (Map<String, Object> maps : s) {
            if (maps != null && maps.containsKey("abcd")) {
                String abcd = maps.get("abcd").toString();
                if(!response.containsKey(abcd))
                    response.put(maps.get("abcd").toString(), maps);
                else {
                    for (Map.Entry<String, Object> entry: maps.entrySet()) {
                        response.get(abcd).put(entry.getKey(), entry.getValue());
                    }
                }
            }
        }
    }
}

来自API的响应：

{ 
   {
    "abcd": 1,
    "cde": 2
   },
   { 
    "abcd": 2,
     "cde": 3
   }
}

将上述响应解析为：

{
   "1" : {
    "abcd": 1,
    "cde": 2
   },
   "2":{ 
    "abcd": 2,
     "cde": 3
   }

}

Answer 1

我认为您的问题在于CompletableFuture.supplyAsync()没有被阻塞，因此您的代码立即向前移动，而不是等待应用异步操作。

代码是以异步方式执行的，在本例中，这意味着告诉CompletableFuture.supplyAsync()中的操作要执行，然后进一步操作。只有当SupplyAsync中的代码完成执行时，才调用thenApply()部分，这很可能发生在您已经返回响应之后。

如果希望在返回响应之前等待所有CompletableFuture.join()方法完成执行。

首先尝试重构代码，去掉theenapply()部分，以便您的~completablefutures`得到部分响应。

然后将所有CompletableFutures分配给一些变量（在我的示例中是MyFirstFuture、MySecondFuture等）。

之后，使用流对所有CompletableFutures应用join()方法，并对每个结果应用putinResponse方法。

例如：

public Map<String, Map<String, Object>> getAllValuesInParallel2(RequestObj requestObj) {

    Map<String, Map<String, Object>> response = new HashMap<>();
    ExecutorService executor = Executors.newFixedThreadPool(6);
    CompletableFuture<List<Map<String, Object>>> myFuture1 = CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("mani");
        return aClient.transform(keys, requestObj);
    }, executor);

    CompletableFuture<List<Map<String, Object>>> myFuture2 = CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("gani");
        return bClient.transform(keys, requestObj);
    }, executor);

    CompletableFuture<List<Map<String, Object>>> myFuture3 = CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("priya");
        return cClient.transform(keys, requestObj);
    }, executor);

    CompletableFuture<List<Map<String, Object>>> myFuture4 = CompletableFuture.supplyAsync(() -> {
        List<Template> keys = new ArrayList<>();
        keys.add("ravi");
        return dClient.transform(keys, requestObj);
    }, executor);

    Stream.of(myFuture1, myFuture2, myFuture3, myFuture4)
        .map(CompletableFuture::join)
        .filter(Objects::nonNull)
        .forEachOrdered(s -> putIntoResponse(response, s));


    return response;
}