ExecutorService exec = Executors.newFixedThreadPool(SOME_NUM_OF_THREADS);
try {
for (final Object o : list) {
exec.submit(new Runnable() {
@Override
public void run() {
// do stuff with o.
}
});
}
} finally {
exec.shutdown();
}
Based on TheLQ's comments, you would set SUM_NUM_THREADS to Runtime.getRuntime().availableProcessors();
编辑: 决定添加一个基本的“并行。为”实现
public class Parallel {
private static final int NUM_CORES = Runtime.getRuntime().availableProcessors();
private static final ExecutorService forPool = Executors.newFixedThreadPool(NUM_CORES * 2, new NamedThreadFactory("Parallel.For"));
public static <T> void For(final Iterable<T> elements, final Operation<T> operation) {
try {
// invokeAll blocks for us until all submitted tasks in the call complete
forPool.invokeAll(createCallables(elements, operation));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static <T> Collection<Callable<Void>> createCallables(final Iterable<T> elements, final Operation<T> operation) {
List<Callable<Void>> callables = new LinkedList<Callable<Void>>();
for (final T elem : elements) {
callables.add(new Callable<Void>() {
@Override
public Void call() {
operation.perform(elem);
return null;
}
});
}
return callables;
}
public static interface Operation<T> {
public void perform(T pParameter);
}
}
并行的用法示例
// Collection of items to process in parallel
Collection<Integer> elems = new LinkedList<Integer>();
for (int i = 0; i < 40; ++i) {
elems.add(i);
}
Parallel.For(elems,
// The operation to perform with each item
new Parallel.Operation<Integer>() {
public void perform(Integer param) {
System.out.println(param);
};
});
// A thread pool which runs for the life of the application.
private static final ExecutorService EXEC =
Executors.newFixedThreadPool(SOME_NUM_OF_THREADS);
//later
EXEC.invokeAll(tasks); // you can optionally specify a timeout.
当使用400万个项目数组进行测试时,这个
为() on 提供5倍于顺序的速度
我的酷睿 i72630QM CPU。
public class Loop {
public interface Each {
void run(int i);
}
private static final int CPUs = Runtime.getRuntime().availableProcessors();
public static void withIndex(int start, int stop, final Each body) {
int chunksize = (stop - start + CPUs - 1) / CPUs;
int loops = (stop - start + chunksize - 1) / chunksize;
ExecutorService executor = Executors.newFixedThreadPool(CPUs);
final CountDownLatch latch = new CountDownLatch(loops);
for (int i=start; i<stop;) {
final int lo = i;
i += chunksize;
final int hi = (i<stop) ? i : stop;
executor.submit(new Runnable() {
public void run() {
for (int i=lo; i<hi; i++)
body.run(i);
latch.countDown();
}
});
}
try {
latch.await();
} catch (InterruptedException e) {}
executor.shutdown();
}
public static void main(String [] argv) {
Loop.withIndex(0, 9, new Loop.Each() {
public void run(int i) {
System.out.println(i*10);
}
});
}
}