对于单生产者单一消费者方案,哪种Java阻塞队列最有效

Question

我正在研究一个标准的Java系统,它对我的​​生产者有严格的时序要求(1/100s ms很重要).

我有一个生产者将东西放在一个阻塞队列中,然后一个消费者拿起那些东西并将其转储到一个文件中.消费者在数据不可用时阻止.

显然,阻塞队列是适当的接口,但是如果我想最小化生产者的成本,我应该选择哪种实际实现？当我把东西放进队列时,我想尽可能少地玩锁定和分配等东西,我不介意消费者是否需要等待更长时间或者更努力地工作.

是否有更快的实现,因为我只有一个消费者和单个生产者？

Answer 1

嗯,确实没有太多的选择.让我来看看列出的子类:

DelayQueue,LinkedBlockingDeque,PriorityBlockingQueue,和SynchronousQueue都为需要额外功能的特殊情况除外; 在这种情况下,它们没有意义.

这使得只有ArrayBlockingQueue和LinkedBlockingQueue.如果你知道如何判断你是否需要a ArrayList或a LinkedList,你可以自己回答这个问题.

注意,在LinkedBlockingQueue"每次插入时动态创建链接节点"; 这可能会推动你走向ArrayBlockingQueue.

Answer 2

您可以使用Java编写延迟敏感系统,但必须注意内存分配,线程之间传递的信息和锁定.您应该编写一些简单的测试来查看这些内容在服务器上花费的时间.(它们根据操作系统和内存架构而有所不同)

如果这样做,您可以在99.99%的时间内获得稳定的操作时间.这不是正确的实时,但它可能足够接近.在交易系统中,使用Java代替C成本可能花费100英镑/天,但是用C/C++而不是Java开发的成本可能远高于此.例如,它给我们的灵活性和它节省的错误数量.

你可以得到相同数量的抖动,你会在C程序中看到同样的抖动.有人称之为"类C"Java.

不幸的是,通过我工作的服务器上的ArrayBlockingQueue在Java中的线程之间传递一个对象需要大约8微秒,我建议你在服务器上测试它.一个简单的测试是在线程之间传递System.nanoTime()并查看它需要多长时间.

ArrayBlockingQueue有一个"功能",它在每个添加的元素上创建一个对象(尽管没有太多理由这样做)所以如果你找到一个不这样做的标准实现,请告诉我.

Answer 3

LinkedBlockingQueue除非存在内存分配延迟,否则将具有O(1)插入成本.非常大的ArrayBlockingQueue将有O(1)插入成本,除非系统由于垃圾收集而被阻止​​; 但是,它会在容量时阻止插入.

即使使用并发垃圾收集,我也不确定您是否应该使用托管语言编写实时系统.

Answer 4

我同意 Andrew Duffy 的评论，即用 java 实现这样一个时间受限的系统可能是一个错误。无论如何，假设您已与 JVM 结合，并且您不希望该队列运行满（即消费者可以处理负载），我认为您可能最好通过自定义实现来满足您的需求，类似于 ArrayBlockingQueue但针对单一生产者/消费者场景进行了缩减/优化。特别是，我喜欢在生产者端旋转以等待空间而不是阻塞的概念。

我参考了 java.concurrent来源作为指导，并起草了这个算法......

它对我来说看起来相当不错，但它完全未经测试，并且在实践中可能不会更快（可能不是革命性的，但我自己做了它:-)。无论如何，我玩得很开心......你能找到一个缺陷吗？

伪代码：

private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final E[] buffer;
private int head = 0
private int tail = 0;

public void put(E e) {
  if (e == null) throw NullPointerException();

  while (buffer[tail] != null) {
    // spin, wait for space...  hurry up, consumer!
    // open Q: would a tight/empty loop be superior here?
    Thread.sleep(1);
  }

  buffer[tail] = e;
  tail = (tail + 1) % buffer.length;

  if (count.getAndIncrement() == 0)  {
    sync(takeLock) { // this is pseudocode -- should be lock/try/finally/unlock
      notEmpty.signal();
    }
  }
}

public E take() {
  sync(takeLock) {
    while (count.get() == 0) notEmpty.await();
  }
  E item = buffer[head];
  buffer[head] = null;
  count.decrement();
  head = (head + 1) % buffer.length;

  return item;
}