Java 7 : How to execute parallel tasks in batches?

Question

我有三个可以并行运行的网络服务调用。因此，我使用 3 个线程的固定池来运行它们。

现在我想处理更多的 Web 服务调用，它们可以并行运行，但前提是前三个调用已处理完毕。

如何批量处理它们？我希望批处理中的那些并行运行。并且每个批处理仅在前一个批处理完成后运行。

到目前为止，我只使用三种服务。如何对它们进行批处理并开始使用另外 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() 方法。它接受一组任务，将它们提交给执行程序并等待直到完成(或直到超时到期)。下面是一个按顺序执行三个批处理的简单示例。每个批处理包含三个并行运行的任务:

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 来检查任务的完成状态(它们是成功执行还是由于异常而终止)，获取它们的返回值等。