Java的"Parallel.For"？

Question

我想知道是否有一个Parallel.For相当于Java的.net版本？

如果有人可以提供一个例子吗？谢谢!

Answer 1

我想最接近的是:

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();
}

根据TheLQ的注释,您可以将SUM_NUM_THREADS设置为 Runtime.getRuntime().availableProcessors();

编辑:决定添加一个基本的"Parallel.For"实现

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);
    }
}

Parallel.For的示例用法

// 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);
    };
});

我想这个实现与Parallel.ForEach更相似

编辑 我把它放在GitHub上,如果有人有兴趣的话.并行For GitHub

Answer 2

MLaw的解决方案是一个非常实用的Parallel.ForEach.我添加了一些修改来制作Parallel.For.

public class Parallel
{
static final int iCPU = Runtime.getRuntime().availableProcessors();

public static <T> void ForEach(Iterable <T> parameters,
                   final LoopBody<T> loopBody)
{
    ExecutorService executor = Executors.newFixedThreadPool(iCPU);
    List<Future<?>> futures  = new LinkedList<Future<?>>();

    for (final T param : parameters)
    {
        Future<?> future = executor.submit(new Runnable()
        {
            public void run() { loopBody.run(param); }
        });

        futures.add(future);
    }

    for (Future<?> f : futures)
    {
        try   { f.get(); }
        catch (InterruptedException e) { } 
        catch (ExecutionException   e) { }         
    }

    executor.shutdown();     
}

public static void For(int start,
                   int stop,
               final LoopBody<Integer> loopBody)
{
    ExecutorService executor = Executors.newFixedThreadPool(iCPU);
    List<Future<?>> futures  = new LinkedList<Future<?>>();

    for (int i=start; i<stop; i++)
    {
        final Integer k = i;
        Future<?> future = executor.submit(new Runnable()
        {
            public void run() { loopBody.run(k); }
        });     
        futures.add(future);
    }

    for (Future<?> f : futures)
    {
        try   { f.get(); }
        catch (InterruptedException e) { } 
        catch (ExecutionException   e) { }         
    }

    executor.shutdown();     
}
}

public interface LoopBody <T>
{
    void run(T i);
}

public class ParallelTest
{
int k;  

public ParallelTest()
{
    k = 0;
    Parallel.For(0, 10, new LoopBody <Integer>()
    {
        public void run(Integer i)
        {
            k += i;
            System.out.println(i);          
        }
    });
    System.out.println("Sum = "+ k);
}

public static void main(String [] argv)
{
    ParallelTest test = new ParallelTest();
}
}

Answer 3

建立在mlaw建议的基础上,添加CountDownLatch.添加chunksize以减少submit().

当使用400万个项目阵列进行测试时,这个在我的Core i7 2630QM CPU上为()的顺序提供了5倍的加速.

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);
            }
        });
    }
}

Answer 4

Java 7中的Fork join框架用于并发支持.但我不知道确切的等价物Parallel.For.

Answer 5

以下是我对此主题的贡献https://github.com/pablormier/parallel-loops.用法很简单:

Collection<String> upperCaseWords = 
    Parallel.ForEach(words, new Parallel.F<String, String>() {
        public String apply(String s) {
            return s.toUpperCase();
        }
    });

也可以更改一些行为方面,例如线程数(默认情况下它使用缓存的线程池):

Collection<String> upperCaseWords = 
            new Parallel.ForEach<String, String>(words)
                .withFixedThreads(4)
                .apply(new Parallel.F<String, String>() {
                    public String apply(String s) {
                        return s.toUpperCase();
                    }
                }).values();

所有代码都是自包含在一个java类中,并且没有比JDK更多的依赖项.我还鼓励您使用Java 8 检查以函数式方式并行化的新方法