ThrottleTaskInvoker.java

package latch.common.utils;

import com.google.common.collect.Lists;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

public class ThrottledTaskInvoker<T> {
    private final ExecutorService executorService;

    public ThrottledTaskInvoker(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public List<T> invoke(List<Callable<T>> tasks, int maxThreadsAtATime) throws ExecutionException, InterruptedException {
        List<List<Callable<T>>> subTasks = Lists.partition(tasks, maxThreadsAtATime);

        List<T> results = Lists.newArrayListWithExpectedSize(tasks.size());
        for (List<Callable<T>> subTask : subTasks) {
            invokeSubTasks(results, subTask);
        }

        return results;
    }

    private void invokeSubTasks(List<T> results, List<Callable<T>> subTasks) throws InterruptedException, ExecutionException {
        List<Future<T>> futures = executorService.invokeAll(subTasks);
        for (Future<T> f : futures) {
            results.add(f.get());
        }
    }
}