CompletableFutureでWebサービスリクエストを並列化する方法は？

Question

私は基本的に入力日付で与えられたデータを要求するサーブレット要求を持っています。複数の日付があるので、複数のリクエストを送信し、結果を集計する必要があります。例：

List<Result> results = new ArrayList<>(); 

for (LocalDate date : dates) { 
    ServletReq req = new ServletReq(date); 

    try { 
     ServletRsp rsp = webservice.send(req); 
     results.addAll(rsp.getResults()); 
    } catch (SpecificException e) { 
     //just ignore this result and continue 
    } 
} 

質問：上記のコードをどのように並列化できますか？手段：複数のServletReq asyncを送信し、結果をリストに収集します。すべての要求が完了するのを待ちます（タイムアウトの可能性もあります）。SpecificExceptionは無視してください。

次のように私が始めたが、これは正しい方向であるかどうかは知りませんか、また、私は完全に上記のコードを転送成功しなかった、どちらも。特に例外に関しては無視されます。

ExecutorService service = Executors.newCachedThreadPool(); 
List<CompletableFuture<ServletRsp>> futures = new ArrayList<>(); 

for (LocalDate date : dates) { 
    ServletReq req = new ServletReq(date); 
    CompletableFuture future = CompletableFuture.supplyAsync(() -> webservice.send(req), service); 
    futures.add(future); 
} 

CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).join(); 

はこれまでのところ、しかし：どのように私は非同期結果にrsp.getResults()を呼び出し、listにすべてを置くことができます。そして、非同期実行中にSpecificExceptionを無視するにはどうすればよいですか？ （私はwebservice.send()の方法を変更できません！）。

Answer 1

• は、サプライヤーの中にそれらをキャッチし、例えば返しますnull。とにかく例外を除いて本当に何もしないのであれば、それをしてください。 future.get()で結果を得るには、nullとExecutionExceptionを処理する必要があります。それら再スロー

例：

CompletableFuture<ServletRsp> future = CompletableFuture.supplyAsync(() -> { 
    try { 
     return webservice.send(new ServletReq(date)); 
    } catch (SpecificException e) { 
     return null; 
    } 
}); 

• （カスタム？）としてRuntimeExceptionので、あなたはそれらを失うことはありません。これで、最後に例外を処理しますが、一部は二重にラップされています。
• 未来を手動で完了してください。

など。

CompletableFuture<ServletRsp> future = new CompletableFuture<>(); 
service.execute(() -> { 
    try { 
     future.complete(webservice.send(new ServletReq(date)); 
    } catch (SpecificException e) { 
     future.completeExceptionally(e); 
    } 
}); 
futures.add(future); 

ExecutionException以外にあるこれ以上のラッピング。 CompletableFuture.supplyAsyncはそれとほぼ同じですが、チェックされた例外を処理するコードはありません。

• だけで投げるコードを受け入れる古き良きExecutorService#submit(Callable<T> callable)メソッドを使用します。

例えば

List<Callable<String>> tasks = dates.stream() 
     .map(d -> (Callable<ServletRsp>)() -> send(new ServletReq(d))) 
     .collect(Collectors.toList()); 
List<Future<ServletRsp>> completed = service.invokeAll(tasks); 

Answer 2

私はそこに良い道があると思います。

問題は、それを自分でやって除き、きちんと結果を収集するためのメカニズムがないこと、である：私はあなたがそれを書いたように、コードの大部分を維持するために

ExecutorService service = Executors.newCachedThreadPool(); 
List<CompletableFuture<Void>> futures = new ArrayList<>(); // these are only references to tell you when the request finishes 
Queue<ServletRsp> results = new ConcurrentLinkedQueue<>(); // this has to be thread-safe 

for (LocalDate date : dates) { 
    ServletReq req = new ServletReq(date); 
    CompletableFuture future = CompletableFuture 
      .supplyAsync(() -> webservice.send(req), service) 
      .thenAcceptAsync(results::add); 
    futures.add(future); 
} 

CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).join(); 
// do stuff with results 

を試してみました。多分それはストリームとビットクリーナーです：

List<CompletableFuture<Void>> collect = dates 
     .map(date -> CompletableFuture 
      .supplyAsync(() -> webservice.send(new ServletReq(date)), service) 
      .thenAcceptAsync(results::add)) 
     .collect(Collectors.toList()); 

    // wait for all requests to finish 
    CompletableFuture.allOf(collect.toArray(new CompletableFuture[collect.size()])).thenAcceptAsync(ignored -> { 
     //you can also handle the response async. 
    });