java - 编写我自己的消费者线程——我没有看到什么？

Question

所以，我有一个 BlockingQueue，我正在用数据(来自许多线程)填充它。我想将这些数据聚合到桶中，比如 1000，然后将它们传递到其他地方。所以我写了一个线程类来轮询队列的末尾，当它有足够的元素时，它会发送聚合数据。

我希望在 java.util.concurrent 中找到一些东西来帮助解决这个问题。我能看到通过 java.util.concurrent 做到这一点的唯一方法是让每次插入队列都添加一个可运行的任务，然后将其添加到聚合集，但这对我来说似乎效率很低。

线程轮询队列策略，假设我有 5 个线程，每个线程可以在本地内存中聚合(顺序并不重要)，然后通过。队列和目的地是竞争的唯一接触点——1 个线程一次可以轮询一个阻塞队列。目的地可能永远不会发生争用。

使用基于任务的方法，使用执行器，所有线程将共享一个聚合点，因此会不断发生争用，更不用说集合的同步/并发变体更慢了。

似乎很明显只有几个线程总是轮询 BlockingQueue。缺点是现在我需要写下它们所有的开始、停止，我需要处理线程死掉的情况，等等。这一切看起来就像我希望在 java.util.concurrent 或者 apache 中找到的样板图书馆。

我真的离预订有那么远吗？一个总是让 x 个线程运行并在它们失败时重新启动它们的类？是否还有另一种我只是没有看到的明显(高效)方法？

Answer 1

试试这个。

public class Consumer<DATA> {

    private List<DATA> dataList = new ArrayList<DATA>();

    private ExecutorService threadPool = Executors.newFixedThreadPool(5);

    public synchronized void consume(DATA data) {

        dataList.add(data);

        if(dataList.size() >= 1000) {

            threadPool.submit(new ConsumerWorker(data));
        }
    }

}

我们实际上是在生产者的线程上下文中累积数据，直到它达到所需的限制。然后我们将这批数据提交到一个线程池，该线程池将根据 ConsumerWorker 的可用性对它们进行排队或执行。您也可以调整线程池的行为。例如，使用 newCachedThreadPool() 将删除非 Activity 线程。