如何共享对象列表以在 Java 中进行并行处理？答案

【问题标题】：How to share a list of objects for parallel processing in Java?如何共享对象列表以在 Java 中进行并行处理？
【发布时间】：2016-05-31 16:41:26
【问题描述】：

我有一个必须在多个线程之间共享的对象列表。

例如，我有 5 个离散对象和一个包含 20 个线程的池。最初，20 个线程中的 5 个将开始工作，其他线程处于 WAIT 状态。 5 个线程将同时使用 5 个对象中的每一个。

如果任何线程完成，它应该释放对象，以便第 6 个线程可以开始工作。

对于线程的并行处理，我想我可以使用 Executor 池。但是如何在线程之间共享对象列表呢？

【问题讨论】：

取一个 BlockingQueue (docs.oracle.com/javase/7/docs/api/java/util/concurrent/…) 并在完成后放回对象
第 6 个线程对对象所做的事情与第 5 个线程不同吗？如果只有 5 个对象，为什么需要 5 个以上的线程。拥有 4 倍所需线程数量当然没有性能优势。
是的。即使它们具有相同的源代码，所有 20 个线程也将具有不同的操作。操作将根据提供给源代码的输入而有所不同。正如 Gilfoy 和 Svetlin Zarev 所建议的，我正在使用一种现在可以完美运行的实现。不过感谢您的 cmets。

标签： java multithreading parallel-processing

【解决方案1】：

我认为您的线程模型错误。目前您想要share，这在并行环境中总是错误的。你应该做的是创建一个pipeline。为此，您有多种选择。

使用（阻塞/并发）队列在线程之间传递对象。在这种情况下，管道的每个阶段都有固定数量的工作线程，它们始终在运行并共享队列。一方面，您有producer 线程，当它们完成put() 对象到队列时，然后consumer 线程之一take() 对象并开始处理它。
使用threadpool 作为来自 (1) 的队列。不是在共享队列上执行put()，而是对共享或专用线程池执行submit()/execute()，以将对象传递到管道的下一个阶段。这种方法的缺点是当前阶段必须知道将任务提交到的确切池，并且还需要知道如何创建要提交的runnable。

PS：在我的回答中，我假设 Thread-6 执行的操作与前 5 个线程不同。如果这个假设不正确 - 解决方案 (1) 仍然正确并且是前进的方向。

【讨论】：

谢谢。有用。你是对的。每个线程将执行不同类型的操作。

【解决方案2】：

在我看来你在谈论Object pool pattern。对象池是包含指定数量对象的容器。当一个对象从池中被取出时，它在池中是不可用的，直到它被放回。

应在以下情况下使用池：

同一对象的高频使用
对象非常大，占用大量内存
对象需要大量时间进行初始化
对象使用大量 IO 操作（流、套接字、数据库等）
对象不是线程安全的

一些出版物不建议使用对象池，尤其是对于只使用内存且不包含外部资源的对象。相关批评question.

取决于你为什么需要这种模式。这里是implementation in Java。

【讨论】：

【解决方案3】：

您可以定义要并行运行的线程数。例如，

ExecutorService executor = Executors.newFixedThreadPool(20);

然后你可以编写一个for循环来处理对象列表。

for(Object obj: objList) {
    Runnable thread = new MyThreadImpl(obj);
    executor.execute(thread);
}

按照您的要求，这应该可以正常工作

【讨论】：

【解决方案4】：

我怀疑 JDK 中是否有标准实现，但您可以根据自己的要求构建自己的对象池，或者可能查看库，例如https://commons.apache.org/proper/commons-pool/

【讨论】：

【解决方案5】：

使用可重入锁

private static synchronized ReentrantLock getLock(Long id) {
    if (!locks.containsKey(id)) {
        locks.put(id, new ReentrantLock());
    }
    return locks.get(id);
}

private void lock(Long id) {
    ReentrantLock lock = getLock(id);
    lock.lock();
}

private void unlock(Long id) {
    ReentrantLock lock = getLock(id);
    if (lock.isHeldByCurrentThread()) {
        lock.unlock();
    }
}

【讨论】：