如何在 Java 中并行处理对象列表

Question

我在 Java 中有一个对象列表，例如 List 中的数千个对象，我正在为每个对象迭代 List 并进行进一步处理。每个对象都在进行相同的处理。这种顺序方法需要很多时间来处理，所以我想用 Java 中的并行处理来实现。我检查了 Java 中的执行程序框架，但我陷入了困境。

我想了一种方法来实现我的要求。

我想实现一些固定数量的最小对象将由每个线程处理，以便每个线程以快速的方式完成其工作和处理对象。我怎样才能做到这一点？或者，如果有任何其他方法可以实现我的要求，请分享。

例如：

列表对象 = new List();

For(Object object : objects) { //对所有的Object做一些通用的操作

}

Answer 1

You can use a ThreadPoolExecutor, it will take care of load balance. Tasks will be distributed on different threads.

Here is an example:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

    public static void main(String[] args) {

        // Fixed thread number
        ExecutorService service = Executors.newFixedThreadPool(10);

        // Or un fixed thread number
        // The number of threads will increase with tasks
        // ExecutorService service = Executors.newCachedThreadPool(10);

        List<Object> objects = new ArrayList<>();
        for (Object o : objects) {
            service.execute(new MyTask(o));
        }

        // shutdown
        // this will get blocked until all task finish
        service.shutdown();
        try {
            service.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static class MyTask implements Runnable {
        Object target;

        public MyTask(Object target) {
            this.target = target;
        }

        @Override
        public void run() {
            // business logic at here
        }
    }
}

Answer 2

有许多并行处理列表的选项：

使用并行流：

objects.stream().parallel().forEach(object -> {
    //Your work on each object goes here, using object
})

如果要使用线程数多于 fork-join 池的池，请使用执行程序服务提交任务：

ExecutorService es = Executors.newFixedThreadPool(10);
for(Object o: objects) {
    es.submit(() -> {
        //code here using Object o...
    }
}

前面的示例与传统的执行程序服务基本相同，在单独的线程上运行任务。

作为这些的替代方案，您还可以使用可完成的未来提交：

//You can also just run a for-each and manually add each
//feature to a list
List<CompletableFuture<Void>> futures = 
    objects.stream().map(object -> CompletableFuture.runAsync(() -> {
    //Your work on each object goes here, using object
})

然后futures，如果需要，您可以使用该对象检查每次执行的状态。