java - java中有没有更好的并行调用api的方法？

Question

我正在使用 Java8 的 java.util.concurrent 的 Callable 接口(interface)调用四个 API。我想要执行多个 Rest 调用，合并结果并返回 JSON。我得到了想要的结果，但需要知道它们是否有更好或更干净的方法来做到这一点。 下面是我的代码。我想知道哪一部分是并行执行的，哪一部分是顺序执行的？

Callable<List<Map<String, Object>>> callable1 = new Callable<List<Map<String, Object>>>()
        {
            @Override
            public List<Map<String, Object>> call() throws Exception
            {  
                return aClient.get();
            }
        };

        Callable<List<Map<String, Object>>> callable2 = new Callable<List<Map<String, Object>>>()
        {
            @Override
            public List<Map<String, Object>> call() throws Exception
            {
                return bClient.get();
            }
        };

        Callable<List<Map<String, Object>>> callable3 = new Callable<List<Map<String, Object>>>()
        {
            @Override
            public List<Map<String, Object>> call() throws Exception
            {
                return cClient.get();
            }
        };

        Callable<List<Map<String, Object>>> callable4 = new Callable<List<Map<String, Object>>>()
        {
            @Override
            public List<Map<String, Object>> call() throws Exception
            {
                return dClient.get()
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(6);
        Future<List<Map<String, Object>>> future1 = executor.submit(callable1);
        Future<List<Map<String, Object>>> future2 = executor.submit(callable2);
        Future<List<Map<String, Object>>> future3 = executor.submit(callable3);
        Future<List<Map<String, Object>>> future4 = executor.submit(callable4);



        Map<String, Map<String, Object>> lists = new HashMap<>();
        try {
            putResult(future1, lists);
            putResult(future2, lists);
            putResult(future3, lists);
            putResult(future4, lists);
        } catch(InterruptedException e) {
            e.printStackTrace();
        } catch(ExecutionException e) {
            e.printStackTrace();
        }
        return metrics;
    }

    private void putResult(Future<List<Map<String, Object>>> future, Map<String, Map<String, Object>> ans) throws InterruptedException, ExecutionException {
        if(future.get() != null && future.get().size() > 0) {
            for (Map<String, Object> maps : future.get()) {
                if (maps != null && maps.containsKey("abcd")) {
                    String abcd = maps.get("abcd").toString();
                    if(!ans.containsKey(abcd))
                        ans.put(maps.get("abcd").toString(), maps);
                    else {
                        for (Map.Entry<String, Object> entry: maps.entrySet()) {
                            ans.get(abcd).put(entry.getKey(), entry.getValue());
                        }
                    }
                }
            }
        }
    }

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

改为

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

}

在解析函数中。请问，用 RxJava 做会更好吗？另外请告诉我如何在 CompletableFutures 中实现上述代码？

Answer 1

如果您使用的是 Java 8+，则可以使用 lambda 来显着缩短代码。

这个:

Callable<List<Map<String, Object>>> callable3 = new Callable<List<Map<String, Object>>>()
{
    @Override
    public List<Map<String, Object>> call() throws Exception
    {
        return cClient.get();
    }
};
Future<List<Map<String, Object>>> future3 = executor.submit(callable3);

可以缩短为:

Future<List<Map<String, Object>>> future3 = executor.submit(cClient::get);

I wanted to know which part is executed parallel and which is sequential?

可调用对象的准备是按顺序进行的。 它们中的每一个的执行都是由执行器并行执行的。 然而，解析代码又是连续的，尽管您没有向我们展示它，所以我不能确定。您也可以使用 future3.thenAccept(/*call to parsing method*/) 或类似函数对其进行并行化，以便执行 API 调用的线程将在完成后解析响应。与工作线程仅返回结果，然后由单个线程解析的实现相比，它应该稍微提高性能。