如何在 Java 中同时处理 Collection 中的元素答案

【问题标题】：How to concurrently process elements in a Collection in Java如何在 Java 中同时处理 Collection 中的元素
【发布时间】：2013-01-23 23:26:30
【问题描述】：

我需要同时处理一些 Collection 实例中的元素。换句话说，而不是迭代 Collection 实例

for (Someclass elem : coll){
     process(elem);
}

我想同时处理这些元素。比如说ConcurrentCollectionExecutor(coll, new Callable{…}, numberOfThreads)。此外，应该修复多个并发线程。

任何灵活的模式已经存在？

【问题讨论】：

将它们添加到队列中，创建固定线程池执行器，从队列中取出，处理并放入其他队列。
在 C# 中它将是一个单行：Parallel.ForEach(coll, process);

标签： java collections concurrency

【解决方案1】：

在名为 MyRunnable 的类中将 process 方法设为 run() 方法，该类实现 Runnable，其构造函数将 elem 作为输入并将其存储为实例变量。然后使用：

ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
for (Someclass elem : coll){
   Runnable worker = new MyRunnable(elem);
   executor.execute(worker);
}

【讨论】：

【解决方案2】：

一个好的解决方案是：

实例化一个包含要处理的元素的ArrayBlockingQueue
实例化 ExecutorService 以同时执行您的处理
实例化你的Runnables，给他们ArrayBlockingQueue作为参数
实现run方法：当队列中有元素时，轮询它们并处理它们
将您的Runnables 提交给ExecutorService

代码：

BlockingQueue<Someclass> toProcess = 
    new ArrayBlockingQueue<Someclass>(coll.size(), false, coll);
ExecutorService es = Executors.newFixedThreadPool(numberOfThreads);
for(int count = 0 ; count < numberOfThreads ; ++c) {
    es.submit(new MyRunnable(toProcess));
}


private static class MyRunnable() implements Runnable {
    private final BlockingQueue<Someclass> toProcess;

    public MyRunnable(BlockingQueue<Someclass> toProcess) {
        this.toProcess = toProcess;
    }

    @Override
    public void run() {
        Someclass element = null;
        while((element = toProcess.poll()) != null) {
            process(element);
        }
    }
}

【讨论】：

我认为这相当于stepanians回答。它只是使用 executorservice 作为任务队列，而不是创建一个明确的任务队列。如果需要将元素添加到队列中，您的会更好。
直接调用 process() 与 run() 中即将内联的 process() 相比不会有很大的性能差异。当有足够的内核或更糟糕的套接字时，扩展受到队列实现的限制。因此，您要确保每个队列轮询或获取值得大约 100 微秒或更多微秒的工作。 Vakimshaars 示例只有在启动线程之前所有作业都存在时才会更好。就像这里的情况一样。

【解决方案3】：

低于此类执行器类的“手工”版本。请注意，您必须传递的不是Callable（或Runnable）的实例，而是此类处理器类的类名。

public class ConcurrentCollectionExecutor<T> {

private Collection<T> collection;
private Class<Runnable> processor;
private int numberOfThreads;
private Executor executor;

public ConcurrentCollectionExecutor(Collection<T> collection, Class<Runnable> processor, int numberOfThreads) {
    this.collection = collection;
    this.processor = processor;
    this.numberOfThreads = numberOfThreads;
    this.executor = Executors.newFixedThreadPool(numberOfThreads);
}

public void run() {
    try {
        Constructor<Runnable> constructor = null;
        for (T t : collection) {
            if (constructor == null) {
                constructor = processor.getConstructor(t.getClass());
            }
            executor.execute(constructor.newInstance(t));
        }
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}      
}

【讨论】：

ey @Andremoniy，我研究了你的代码，发现它非常有趣......但是我想添加一些东西来完善它......如果你允许我，请。 .. 1) 可运行实现必须具有一个构造函数，该构造函数将与数组的组件类型相同类型的元素作为唯一参数。 2）您应该添加并结束运行executor.shutdow() 3）未使用实例字段numberOfThreads。希望贡献和问候！

【解决方案4】：

我不知道这有什么模式，但作为一个想法，你可以根据线程数来划分你的集合元素，所以每个线程都有 X 个元素来处理，例如：

Collection 有 20 个元素，你所有的函数都提供 4 个线程，然后实习生像这样开始它们：

thread1 gets the elements [0 .. 4]
thread2 gets the elements [5 .. 9]
thread3 gets the elements [10 .. 14]
thread1 gets the elements [15 .. 19]

请注意，从集合中删除元素可能会导致问题，然后线程 4 会在您的集合中的元素少于 20 个时尝试访问元素 [19]。

编辑：

正如大脑提到的取决于元素处理时间，这个想法可能不是有效的，好像处理前 5 个元素中的一个需要 10 秒，但其他元素只需要 0.5 秒，然后线程 1 会很忙，但其他线程会最终不会长时间并行运行。

【讨论】：

我认为这比其他两种解决方案弱，因为处理不同的元素可能需要不同的时间长度。例如。如果处理前 5 个元素中的一个需要 10 秒，但其他元素只需要 0.5 秒，那么线程 1 会很忙，但其他线程最终不会并行运行很长时间。
@brain 这是一个很好的观点，我会写在答案中，谢谢。