有没有一种简单的方法可以并行化Java中的foreach循环? [英] Is there a easy way to parallelize a foreach loop in java?

问题描述

是否有一种简单的方法可以使用某些库内容并行化Java 8中的foreach循环?

Is there a easy way to parallelise a foreach loop in java 8 using some library stuff?

void someFunction(SomeType stuff, SomeType andStuff) {
    for (Object object : lotsOfObjects)
        object.doSomethingThatCanBeDoneInParallel(stuff, andStuff);
}

多线程是一个痛苦且耗时的过程，所以我想知道是否存在使用某些库完成上述操作的更简单方法.

Multithreading is kinda painful and time consuming so i wonder if there is a simpler way to do the above using some library.

谢谢.

于2018年6月3日编辑

edited in 3/06/2018

ExecutorServices确实非常方便，我不能使用shutdown()来等待，因为我每帧都运行一次东西，并且每帧创建一个新的ExecutorServices太昂贵了.

ExecutorServices is very handy indeed, I can't use shutdown() to wait because I run the thing every frame and create a new ExecutorServices every frame would be too expensive.

我最终写了一个类来并行化一个fori循环，我以为我可以和其他像我这样的新手分享它.

I ended up writing a class to parallelize a fori loop and I thought I share it with other newbies like me.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

public class ParallelForI {
    public ParallelForI(int numberOfThread) {
        NUMBER_OF_THREAD = numberOfThread;
        executorService = Executors.newFixedThreadPool(NUMBER_OF_THREAD);
        finished = new AtomicBoolean[NUMBER_OF_THREAD];
        for (int i = 0; i < finished.length; i++)
            finished[i] = new AtomicBoolean(true);
        // true is better for waitForLastRun before any run.
    }
    private ExecutorService executorService;
    private final int NUMBER_OF_THREAD;

    private AtomicBoolean[] finished;
    public void waitForLastRun() {
        synchronized (this) {
        /* synchronized outside the loop so other thread
         can't notify when it's not waiting. */
            for (int i = 0; i < NUMBER_OF_THREAD; i++) {
                if (!finished[i].get()) {
                    i = -1;
                    try {
                        this.wait(); //
                    } catch (InterruptedException e) {
                        // do nothing and move one.
                    }
                }
            }
        }
    }

    public void run(FunctionForI functionForI, final int MAX_I) {
        for (AtomicBoolean finished : finished)
            finished.set(false); // just started
        for (int i = 0; i < NUMBER_OF_THREAD; i++) {
            final int threadNumber = i;
            executorService.submit(new Runnable() {
                @Override // use lambda if you have java 8 or above
                public void run() {
                    int iInitial = threadNumber * MAX_I / NUMBER_OF_THREAD;
                    int iSmallerThan;
                    if (threadNumber == NUMBER_OF_THREAD - 1) // last thread
                        iSmallerThan = MAX_I;
                    else
                        iSmallerThan = (threadNumber + 1) * MAX_I / NUMBER_OF_THREAD;
                    for (int i1 = iInitial; i1 < iSmallerThan; i1++) {
                        functionForI.run(i1);
                    }
                    finished[threadNumber].set(true);
                    synchronized (this) {
                        this.notify();
                    }
                }
            });
        }
    }

    public interface FunctionForI {
        void run(int i);
    }
}

这是使用它的方式:

void someFunction(final SomeType stuff, final SomeType andStuff) {
    ParallelForI parallelForI = new parallelForI(numberOfThread);
    // swap numberOfThread with a suitable int

    parallelForI.run(new ParallelForI.FunctionForI() {
        @Override // use lambda if you have java 8 or above
        public void run(int i) {
            lotsOfObjects[i].doSomethingThatCanBeDoneInParallel(stuff, andStuff);
            // don't have to be array.
        }
    }, lotsOfObjects.length); // again, don't have to be array

    parallellForI.waitForLastRun(); // put this where ever you want 
    // You can even put this before parallelForI.run().
    // Although it doesn't make sense to do that...
    // Unlike shutdown() waitForLastRun() will not cause parallelForI to reject future task.
}

推荐答案

解决方案可能是在Thread中启动每个任务，如下所示:

A solution could be to launch every task in a Thread as follows:

new Thread(() -> object.doSomethingThatCanBeDoneInParallel(stuff, andStuff)).start();

但这不是一个相关的解决方案，因为创建线程的成本很高，因此有一些机制和工具可以帮助您:Executors类用于构建某些pools.

but this is not a relevant solution as Thread creation is costly, so there are mechanisms and tools to help you: the Executors class to build some pools.

一旦有了将要管理该实例的实例，便会为其提供任务，这些任务将根据您选择的线程数并行运行:

Once you have the instance that will manage this, you provide it with tasks, which will run in parallel, on the number of threads you choose:

void someFunction(SomeType stuff, SomeType andStuff) {
    ExecutorService exe = Executors.newFixedThreadPool(4);   // 4 can be changed of course
    for (Object object : lotsOfObjects) {
        exe.submit(() -> object.doSomethingThatCanBeDoneInParallel(stuff, andStuff));
    }

    // Following lines are optional, depending if you need to wait until all tasks are finished or not
    exe.shutdown();
    try {
        exe.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }   
}

这篇关于有没有一种简单的方法可以并行化Java中的foreach循环?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！