@Test(expectedExceptions = ExecutionException.class) public void testParallelizeTimeout() throws Exception { List<Integer> nums = ImmutableList.of(3, 5); final int factor = 5; Function<Integer, String> sleepAndMultiply = new Function<Integer, String>() { @Override public String apply(Integer input) { try { if (input == 5) { TimeUnit.SECONDS.sleep(2); } } catch (InterruptedException e) { throw new RuntimeException(e); } return Integer.toString(input * factor); } }; ExecutorsUtils.parallelize(nums, sleepAndMultiply, 2, 1, Optional.<Logger> absent()); }
public static <F, T> List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount, int timeoutInSecs, Optional<Logger> logger) throws ExecutionException { Preconditions.checkArgument(list != null, "Input list can not be null"); Preconditions.checkArgument(function != null, "Function can not be null"); final List<T> results = Lists.newArrayListWithCapacity(list.size()); List<Future<T>> futures = Lists.newArrayListWithCapacity(list.size()); ExecutorService executorService = MoreExecutors.getExitingExecutorService( (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount, ExecutorsUtils.newThreadFactory(logger)), 2, TimeUnit.MINUTES); for (final F l : list) { futures.add(executorService.submit(new Callable<T>() { @Override public T call() throws Exception { return function.apply(l); } })); } ExecutorsUtils.shutdownExecutorService(executorService, logger, timeoutInSecs, TimeUnit.SECONDS); for (Future<T> future : futures) { try { results.add(future.get()); } catch (InterruptedException e) { throw new ExecutionException("Thread interrupted", e); } } return results; }
ExecutorsUtils { public static <F, T> List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount, int timeoutInSecs, Optional<Logger> logger) throws ExecutionException { Preconditions.checkArgument(list != null, "Input list can not be null"); Preconditions.checkArgument(function != null, "Function can not be null"); final List<T> results = Lists.newArrayListWithCapacity(list.size()); List<Future<T>> futures = Lists.newArrayListWithCapacity(list.size()); ExecutorService executorService = MoreExecutors.getExitingExecutorService( (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount, ExecutorsUtils.newThreadFactory(logger)), 2, TimeUnit.MINUTES); for (final F l : list) { futures.add(executorService.submit(new Callable<T>() { @Override public T call() throws Exception { return function.apply(l); } })); } ExecutorsUtils.shutdownExecutorService(executorService, logger, timeoutInSecs, TimeUnit.SECONDS); for (Future<T> future : futures) { try { results.add(future.get()); } catch (InterruptedException e) { throw new ExecutionException("Thread interrupted", e); } } return results; } }
ExecutorsUtils { public static <F, T> List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount, int timeoutInSecs, Optional<Logger> logger) throws ExecutionException { Preconditions.checkArgument(list != null, "Input list can not be null"); Preconditions.checkArgument(function != null, "Function can not be null"); final List<T> results = Lists.newArrayListWithCapacity(list.size()); List<Future<T>> futures = Lists.newArrayListWithCapacity(list.size()); ExecutorService executorService = MoreExecutors.getExitingExecutorService( (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount, ExecutorsUtils.newThreadFactory(logger)), 2, TimeUnit.MINUTES); for (final F l : list) { futures.add(executorService.submit(new Callable<T>() { @Override public T call() throws Exception { return function.apply(l); } })); } ExecutorsUtils.shutdownExecutorService(executorService, logger, timeoutInSecs, TimeUnit.SECONDS); for (Future<T> future : futures) { try { results.add(future.get()); } catch (InterruptedException e) { throw new ExecutionException("Thread interrupted", e); } } return results; }  }
ExecutorsUtils { public static <F, T> List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount, int timeoutInSecs, Optional<Logger> logger) throws ExecutionException { Preconditions.checkArgument(list != null, "Input list can not be null"); Preconditions.checkArgument(function != null, "Function can not be null"); final List<T> results = Lists.newArrayListWithCapacity(list.size()); List<Future<T>> futures = Lists.newArrayListWithCapacity(list.size()); ExecutorService executorService = MoreExecutors.getExitingExecutorService( (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount, ExecutorsUtils.newThreadFactory(logger)), 2, TimeUnit.MINUTES); for (final F l : list) { futures.add(executorService.submit(new Callable<T>() { @Override public T call() throws Exception { return function.apply(l); } })); } ExecutorsUtils.shutdownExecutorService(executorService, logger, timeoutInSecs, TimeUnit.SECONDS); for (Future<T> future : futures) { try { results.add(future.get()); } catch (InterruptedException e) { throw new ExecutionException("Thread interrupted", e); } } return results; }  static ThreadFactory defaultThreadFactory(); static ThreadFactory newThreadFactory(Optional<Logger> logger); static ThreadFactory newThreadFactory(Optional<Logger> logger, Optional<String> nameFormat); static ThreadFactory newDaemonThreadFactory(Optional<Logger> logger, Optional<String> nameFormat); static ListeningExecutorService loggingDecorator(ExecutorService executorService); static ListeningScheduledExecutorService loggingDecorator(ScheduledExecutorService scheduledExecutorService); static Runnable loggingDecorator(Runnable runnable); static Callable<T> loggingDecorator(Callable<T> callable); static void shutdownExecutorService(ExecutorService executorService, Optional<Logger> logger, long timeout,
      TimeUnit unit); static void shutdownExecutorService(ExecutorService executorService, Optional<Logger> logger); static List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount,
      int timeoutInSecs, Optional<Logger> logger); }
ExecutorsUtils { public static <F, T> List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount, int timeoutInSecs, Optional<Logger> logger) throws ExecutionException { Preconditions.checkArgument(list != null, "Input list can not be null"); Preconditions.checkArgument(function != null, "Function can not be null"); final List<T> results = Lists.newArrayListWithCapacity(list.size()); List<Future<T>> futures = Lists.newArrayListWithCapacity(list.size()); ExecutorService executorService = MoreExecutors.getExitingExecutorService( (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount, ExecutorsUtils.newThreadFactory(logger)), 2, TimeUnit.MINUTES); for (final F l : list) { futures.add(executorService.submit(new Callable<T>() { @Override public T call() throws Exception { return function.apply(l); } })); } ExecutorsUtils.shutdownExecutorService(executorService, logger, timeoutInSecs, TimeUnit.SECONDS); for (Future<T> future : futures) { try { results.add(future.get()); } catch (InterruptedException e) { throw new ExecutionException("Thread interrupted", e); } } return results; }  static ThreadFactory defaultThreadFactory(); static ThreadFactory newThreadFactory(Optional<Logger> logger); static ThreadFactory newThreadFactory(Optional<Logger> logger, Optional<String> nameFormat); static ThreadFactory newDaemonThreadFactory(Optional<Logger> logger, Optional<String> nameFormat); static ListeningExecutorService loggingDecorator(ExecutorService executorService); static ListeningScheduledExecutorService loggingDecorator(ScheduledExecutorService scheduledExecutorService); static Runnable loggingDecorator(Runnable runnable); static Callable<T> loggingDecorator(Callable<T> callable); static void shutdownExecutorService(ExecutorService executorService, Optional<Logger> logger, long timeout,
      TimeUnit unit); static void shutdownExecutorService(ExecutorService executorService, Optional<Logger> logger); static List<T> parallelize(final List<F> list, final Function<F, T> function, int threadCount,
      int timeoutInSecs, Optional<Logger> logger); static final long EXECUTOR_SERVICE_SHUTDOWN_TIMEOUT; static final TimeUnit EXECUTOR_SERVICE_SHUTDOWN_TIMEOUT_TIMEUNIT; }