Java 7：並列タスクをバッチで実行する方法は？

Question

私は3つのWebサービスコールを並行して実行できます。したがって、私はそれらを実行する3スレッドの固定プールを使用しています。

ここでは、最初の3つの呼び出しが処理された後にのみ、並列に実行できる2つ以上のWebサービス呼び出しを処理します。

どうすればそれらをバッチすることができますか？バッチ内のものを並行して実行したい。また、すべてのバッチは、前のバッチが完了した後にのみ実行されます。

これまでのところ、私は3つのサービスしか扱っていません。どのようにそれらをバッチし、別の2つのサービスを使い始めることができますか？

ExecutorService peopleDataTaskExecutor = Executors.newFixedThreadPool(3); 

    Future<Collection<PeopleInterface>> task1 = null; 
    if (condition) { 
     task1 = peopleDataTaskExecutor.submit(buildTask1Callable(mycontext)); 
    } 

    Future<Map<String, Task2Response>> task2 = peopleDataTaskExecutor.submit(buildTask2Callable(mycontext)); 

    Future<Map<String, Task3Response>> task3 = null; 
    task3 = peopleDataTaskExecutor.submit(buildTask3Callable(mycontext)); 

    peopleDataTaskExecutor.shutdown(); 
    try { 
     peopleDataTaskExecutor.awaitTermination(10, TimeUnit.SECONDS); 
    } catch (InterruptedException e) { 
    } 

    Collection<PeopleInterface> task1Data = null; 
    try { 
     task1Data = task1 != null ? task1.get() : null; 
    } catch (InterruptedException | ExecutionException e) { 
    } 

    Map<String, Task2Response> task2Data = null; 
    try { 
     task2Data = task2.get(); 
    } catch (InterruptedException | ExecutionException e) { 
    } 

    Map<String, Task3Response> task3Data = null; 
    if (task3 != null) { 
     try { 
      task3Data = task3.get(); 
     } catch (InterruptedException | ExecutionException e) { 
     } 
    } 

Answer 1

バッチを順番に実行する最も簡単な方法は、invokeAll()メソッドを使用することです。タスクのコレクションを受け取り、実行プログラムに提出し、完了するまで（またはタイムアウトが切れるまで）待ちます。ここでは、3つのバッチを順番に実行する簡単な例を示します。各バッチは、並列に実行されている3つのタスクが含まれています

public class Program { 
    static class Task implements Callable<Integer> { 
     private static Random rand = new Random(); 
     private final int no; 
     Task(int no) { 
      this.no = no; 
     } 
     @Override 
     public Integer call() throws Exception { 
      Thread.sleep(rand.nextInt(5000)); 
      System.out.println("Task " + no + " finished"); 
      return no; 
     } 
    } 

    public static void main(String[] args) throws Exception { 
     ExecutorService executor = Executors.newFixedThreadPool(3); 
     processBatch(executor, 1); 
     processBatch(executor, 2); 
     processBatch(executor, 3); 
     executor.shutdown(); 
    } 

    private static void processBatch(ExecutorService executor, int batchNo) throws InterruptedException { 
     Collection batch = new ArrayList<>(); 
     batch.add(new Task(batchNo * 10 + 1)); 
     batch.add(new Task(batchNo * 10 + 2)); 
     batch.add(new Task(batchNo * 10 + 3)); 
     List<Future> futures = executor.invokeAll(batch); 
     System.out.println("Batch " + batchNo + " proceseed"); 
    } 
} 

あなたは彼らのリターンを得る、（彼らが正常に実行されるか、例外のため、終了しました）タスクの完了状態を確認するためにprocessBatch()方法でこれらのFuture Sを使用することができます値など