java - Java中的生产者/消费者 : keeping it balanced

Question

我有一个 Java 文本处理应用程序，它逐 block 读取文件(约 100000 行)并在单独的线程中处理每个 block 。

效果很好，但有一个问题。读取行比处理行快得多，并且程序最终会形成一个等待轮到的 Runnable 队列。这会消耗一些内存，我打算节省这些内存。

我希望程序这样做:

• 读取 16 个 block 并将它们提交给 8 个可运行对象；
• 如果未处理的 block 数低于 12，则再读取 4 个文本 block 。

这将使 Runnables 保持忙碌，但同时保留用于处理的内存(而不是存储 block )。

如何在 Java 中做到这一点？用 preudocode 编写我想要这个:

loop {

  chunk = readChunkOfData();

  counter.inc();    

  processAsync(chunk);

  if (counter.isBiggerThan(16)) {
    counter.sleepWhileCounterIsBiggerThan(12);
  }
}

...

worker {
  // do the job

  counter.dec();
}

Answer 1

正如 Marko Topolnik 评论的那样，使用有界(阻塞)queues可以优雅地解决你的问题。

您不需要计数器，因为队列知道其限制，并且您的伪代码最终将类似于以下内容

loop {
    chunk = readChunkOfData();
    queue.put(chunk);
}

worker {
    chunk = queue.take();
    process(chunk);
}

这假设队列是例如 new ArrayBlockingQueue(16);并为全体职工所共享。您还可以使用drainTo(Collection<? super E> c, int maxElements)在工作人员中一次获取多个 block ，作为工作人员一侧的额外工作缓冲区，但这可能不会产生太大的区别。