如何在Java中并发处理集合中的元素

Question

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

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

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

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

Answer 1

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

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

Answer 2

一个好的解决方案是：

实例化一个包含要执行处理的元素的ArrayBlockingQueue来执行处理，您的ExecutorService将这些元素作为concurrently

• instantiate

• 方法提供给它们:当队列中有元素时，轮询它们并对它们进行处理，然后向ExecutorService

提交您的<Runnable>d15>s

代码：

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

Answer 3

下面是这个executor类的“手工”版本。请注意，您必须传递的不是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);
    }
}      
}