java - 有没有一种简单的方法可以在java中并行化foreach循环?
问题描述
有没有一种简单的方法可以使用一些库的东西来并行化 java 8 中的 foreach 循环?
void someFunction(SomeType stuff, SomeType andStuff) {
for (Object object : lotsOfObjects)
object.doSomethingThatCanBeDoneInParallel(stuff, andStuff);
}
多线程有点痛苦和耗时,所以我想知道是否有更简单的方法可以使用一些库来完成上述操作。
谢谢。
编辑于 2018 年 3 月 6 日
ExecutorServices 确实非常方便,我不能使用 shutdown() 等待,因为我每帧都运行这个东西并每帧创建一个新的 ExecutorServices 太昂贵了。
我最终编写了一个类来并行化一个 fori 循环,我想我会与像我这样的其他新手分享它。
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.
}
解决方案
一个解决方案可能是在 a 中启动每个任务,Thread
如下所示:
new Thread(() -> object.doSomethingThatCanBeDoneInParallel(stuff, andStuff)).start();
但这不是一个相关的解决方案,因为创建线程的成本很高,因此有一些机制和工具可以帮助您:Executors
构建一些pools
.
一旦你有了管理这个的实例,你就可以为它提供任务,这些任务将在你选择的线程数上并行运行:
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();
}
}
推荐阅读
- python - Python中的图像创建和字体
- r - R函数定义困难
- python - 如何在单独的父文件夹中读取 csv 文件?
- react-native - Android 设备上的 OnScroll 动画不流畅
- python - 寻求帮助将大字符串变成许多较小的变量
- laravel - Laravel 网站翻译强制语言环境
- plugins - Post 操作 Update 中的 PreImage null
- php - laravel php artisan migrate 错误在 ProviderRepository.php 第 208 行:找不到类“Facade\Ignition\IgnitionServiceProvider”
- c++ - boost flat_map 批量插入
- python - 如何检查和选择第一个工作链接?