@Test public void parallelRunEachAction() { final Action1<Integer> action1 = new Action1<Integer>() { @Override public void perform(Integer x) { for (int i = LowerBound; i < UpperBound; i++) { Hero.findRoot(i); } if (log.isDebugEnabled()) log.debug("FindRoot({}) returns [{}]", UpperBound, Hero.findRoot(UpperBound)); } }; @Cleanup AutoStopwatch stopwatch = new AutoStopwatch(); Parallels.runEach(NumberRange.range(0, 100), action1); }
public static <T> void runEach(final Iterable<T> elements, final Action1<T> action) { shouldNotBeNull(elements, "elements"); shouldNotBeNull(action, "function"); if (isDebugEnabled) log.debug("병렬로 작업을 수행합니다... workerCount=[{}]", getWorkerCount()); ExecutorService executor = Executors.newFixedThreadPool(getWorkerCount()); try { List<T> elemList = Lists.newArrayList(elements); int partitionSize = getPartitionSize(elemList.size(), getWorkerCount()); Iterable<List<T>> partitions = Iterables.partition(elemList, partitionSize); List<Callable<Void>> tasks = Lists.newLinkedList(); for (final List<T> partition : partitions) { Callable<Void> task = new Callable<Void>() { @Override public Void call() throws Exception { for (final T element : partition) action.perform(element); return null; } }; tasks.add(task); } List<Future<Void>> results = executor.invokeAll(tasks); for (Future<Void> result : results) { result.get(); } if (isDebugEnabled) log.debug("모든 작업을 병렬로 수행하였습니다. workerCount=[{}]", getWorkerCount()); } catch (Exception e) { log.error("데이터에 대한 병렬 작업 중 예외가 발생했습니다.", e); throw new RuntimeException(e); } finally { executor.shutdown(); } }
Parallels { public static <T> void runEach(final Iterable<T> elements, final Action1<T> action) { shouldNotBeNull(elements, "elements"); shouldNotBeNull(action, "function"); if (isDebugEnabled) log.debug("병렬로 작업을 수행합니다... workerCount=[{}]", getWorkerCount()); ExecutorService executor = Executors.newFixedThreadPool(getWorkerCount()); try { List<T> elemList = Lists.newArrayList(elements); int partitionSize = getPartitionSize(elemList.size(), getWorkerCount()); Iterable<List<T>> partitions = Iterables.partition(elemList, partitionSize); List<Callable<Void>> tasks = Lists.newLinkedList(); for (final List<T> partition : partitions) { Callable<Void> task = new Callable<Void>() { @Override public Void call() throws Exception { for (final T element : partition) action.perform(element); return null; } }; tasks.add(task); } List<Future<Void>> results = executor.invokeAll(tasks); for (Future<Void> result : results) { result.get(); } if (isDebugEnabled) log.debug("모든 작업을 병렬로 수행하였습니다. workerCount=[{}]", getWorkerCount()); } catch (Exception e) { log.error("데이터에 대한 병렬 작업 중 예외가 발생했습니다.", e); throw new RuntimeException(e); } finally { executor.shutdown(); } } }
Parallels { public static <T> void runEach(final Iterable<T> elements, final Action1<T> action) { shouldNotBeNull(elements, "elements"); shouldNotBeNull(action, "function"); if (isDebugEnabled) log.debug("병렬로 작업을 수행합니다... workerCount=[{}]", getWorkerCount()); ExecutorService executor = Executors.newFixedThreadPool(getWorkerCount()); try { List<T> elemList = Lists.newArrayList(elements); int partitionSize = getPartitionSize(elemList.size(), getWorkerCount()); Iterable<List<T>> partitions = Iterables.partition(elemList, partitionSize); List<Callable<Void>> tasks = Lists.newLinkedList(); for (final List<T> partition : partitions) { Callable<Void> task = new Callable<Void>() { @Override public Void call() throws Exception { for (final T element : partition) action.perform(element); return null; } }; tasks.add(task); } List<Future<Void>> results = executor.invokeAll(tasks); for (Future<Void> result : results) { result.get(); } if (isDebugEnabled) log.debug("모든 작업을 병렬로 수행하였습니다. workerCount=[{}]", getWorkerCount()); } catch (Exception e) { log.error("데이터에 대한 병렬 작업 중 예외가 발생했습니다.", e); throw new RuntimeException(e); } finally { executor.shutdown(); } } private  Parallels(); }
Parallels { public static <T> void runEach(final Iterable<T> elements, final Action1<T> action) { shouldNotBeNull(elements, "elements"); shouldNotBeNull(action, "function"); if (isDebugEnabled) log.debug("병렬로 작업을 수행합니다... workerCount=[{}]", getWorkerCount()); ExecutorService executor = Executors.newFixedThreadPool(getWorkerCount()); try { List<T> elemList = Lists.newArrayList(elements); int partitionSize = getPartitionSize(elemList.size(), getWorkerCount()); Iterable<List<T>> partitions = Iterables.partition(elemList, partitionSize); List<Callable<Void>> tasks = Lists.newLinkedList(); for (final List<T> partition : partitions) { Callable<Void> task = new Callable<Void>() { @Override public Void call() throws Exception { for (final T element : partition) action.perform(element); return null; } }; tasks.add(task); } List<Future<Void>> results = executor.invokeAll(tasks); for (Future<Void> result : results) { result.get(); } if (isDebugEnabled) log.debug("모든 작업을 병렬로 수행하였습니다. workerCount=[{}]", getWorkerCount()); } catch (Exception e) { log.error("데이터에 대한 병렬 작업 중 예외가 발생했습니다.", e); throw new RuntimeException(e); } finally { executor.shutdown(); } } private  Parallels(); static ExecutorService createExecutor(); static ExecutorService createExecutor(int threadCount); static void run(int count, final Runnable runnable); static void run(int fromInclude, int toExclude, final Runnable action); static void run(int fromInclude, int toExclude, int step, final Runnable runnable); static void run(int count, final Action1<Integer> action); static void run(int fromInclude, int toExclude, final Action1<Integer> action); static void run(int fromInclude, int toExclude, int step, final Action1<Integer> action); static List<V> run(int count, final Callable<V> callable); static List<V> run(int fromInclude, int toExclude, final Callable<V> callable); static List<V> run(int fromInclude, int toExclude, int step, final Callable<V> callable); static List<V> run(int count, final Function1<Integer, V> function); static List<V> run(int fromInclude, int toExclude, final Function1<Integer, V> function); static List<V> run(int fromInclude, int toExclude, int step, final Function1<Integer, V> function); static void runEach(final Iterable<T> elements, final Action1<T> action); static List<V> runEach(final Iterable<T> elements, final Function1<T, V> function); static void runPartitions(int count, final Action1<List<Integer>> action); static void runPartitions(int fromInclude, int toExclude, final Action1<List<Integer>> action); static void runPartitions(int fromInclude, int toExclude, int step, final Action1<List<Integer>> action); static List<V> runPartitions(int count, final Function1<List<Integer>, List<V>> function); static List<V> runPartitions(int fromInclude, int toExclude, final Function1<List<Integer>, List<V>> function); static List<V> runPartitions(int fromInclude, int toExclude, int step, final Function1<List<Integer>, List<V>> function); static void runPartitions(final Iterable<T> elements, final Action1<List<T>> action); static List<V> runPartitions(final Iterable<T> elements, final Function1<List<T>, List<V>> function); }
Parallels { public static <T> void runEach(final Iterable<T> elements, final Action1<T> action) { shouldNotBeNull(elements, "elements"); shouldNotBeNull(action, "function"); if (isDebugEnabled) log.debug("병렬로 작업을 수행합니다... workerCount=[{}]", getWorkerCount()); ExecutorService executor = Executors.newFixedThreadPool(getWorkerCount()); try { List<T> elemList = Lists.newArrayList(elements); int partitionSize = getPartitionSize(elemList.size(), getWorkerCount()); Iterable<List<T>> partitions = Iterables.partition(elemList, partitionSize); List<Callable<Void>> tasks = Lists.newLinkedList(); for (final List<T> partition : partitions) { Callable<Void> task = new Callable<Void>() { @Override public Void call() throws Exception { for (final T element : partition) action.perform(element); return null; } }; tasks.add(task); } List<Future<Void>> results = executor.invokeAll(tasks); for (Future<Void> result : results) { result.get(); } if (isDebugEnabled) log.debug("모든 작업을 병렬로 수행하였습니다. workerCount=[{}]", getWorkerCount()); } catch (Exception e) { log.error("데이터에 대한 병렬 작업 중 예외가 발생했습니다.", e); throw new RuntimeException(e); } finally { executor.shutdown(); } } private  Parallels(); static ExecutorService createExecutor(); static ExecutorService createExecutor(int threadCount); static void run(int count, final Runnable runnable); static void run(int fromInclude, int toExclude, final Runnable action); static void run(int fromInclude, int toExclude, int step, final Runnable runnable); static void run(int count, final Action1<Integer> action); static void run(int fromInclude, int toExclude, final Action1<Integer> action); static void run(int fromInclude, int toExclude, int step, final Action1<Integer> action); static List<V> run(int count, final Callable<V> callable); static List<V> run(int fromInclude, int toExclude, final Callable<V> callable); static List<V> run(int fromInclude, int toExclude, int step, final Callable<V> callable); static List<V> run(int count, final Function1<Integer, V> function); static List<V> run(int fromInclude, int toExclude, final Function1<Integer, V> function); static List<V> run(int fromInclude, int toExclude, int step, final Function1<Integer, V> function); static void runEach(final Iterable<T> elements, final Action1<T> action); static List<V> runEach(final Iterable<T> elements, final Function1<T, V> function); static void runPartitions(int count, final Action1<List<Integer>> action); static void runPartitions(int fromInclude, int toExclude, final Action1<List<Integer>> action); static void runPartitions(int fromInclude, int toExclude, int step, final Action1<List<Integer>> action); static List<V> runPartitions(int count, final Function1<List<Integer>, List<V>> function); static List<V> runPartitions(int fromInclude, int toExclude, final Function1<List<Integer>, List<V>> function); static List<V> runPartitions(int fromInclude, int toExclude, int step, final Function1<List<Integer>, List<V>> function); static void runPartitions(final Iterable<T> elements, final Action1<List<T>> action); static List<V> runPartitions(final Iterable<T> elements, final Function1<List<T>, List<V>> function);  }