在 Java 中如何更改 ArrayBlockedQueue 的长度？

Question

我正在为 Android 创建一个媒体播放器应用。我有两个线程：一个产生音频帧，另一个使用这些帧。

我希望我的客户能够尝试使用不同大小的 ArrayBlockedQueue，从“无”缓冲（实际上是 1 个）到最多 10 个缓冲块。

我似乎在 Java 中找不到任何类提供与 ArrayBlockedQueue 类似的功能，但允许我动态地使项目列表更长/更短。

问题 1) 有谁知道功能类似于 ArrayBlockedQueue 的类，但允许我更改要持有的项目数量？

然后我有一个奇怪的想法：我可以捏造它吗？我可以创建一个具有新大小的新 ArrayBlockedQueue，并逐步复制当前在旧 ArrayBlockedQueue 中的 1-10 个项目并将它们放入新的 ArrayBlockedQueue，然后在旧的上存储指向新 ArrayBlockedQueue 的指针吗？

因为永远不会超过 10 个（或者我的缓冲区限制是多少），所以将项目复制到新数组不会花费太多时间。

问题 2）这是一种“合理”的方式来实现 ArrayBlockedQueue 实现，仍然给我灵活性？

问题 3）有没有更好的方法来解决这个问题？

-肯

Answer 1

您可能需要创建自己的 BlockingQueue 实现来包装旧队列和新队列 - 从旧队列轮询直到它为空，然后将其设置为 null 以防止任何内存泄漏。这样您就不会丢失旧队列中任何待处理的puts

MyBlockingQueue {
  private MyBlockingQueue oldQueue
  private ArrayBlockingQueue newQueue

  ArrayBlockingQueue(int newCapacity, MyBlockingQueue _oldQueue) {
    oldQueue = _oldQueue
    newQueue = new ArrayBlockingQueue(newCapacity)
    E oldVal = null
    while(newQueue.remainingCapacity() > 0 && 
         (oldVal = oldPoll) != null)
      newQueue.put(oldVal)
  }

  boolean isEmpty() {
    (oldQueue == null || oldQueue.isEmpty) && newQueue.isEmpty 
  }

  void put(E e) {
    newQueue.put(e)
  }

  E take() {
    E oldVal = oldPoll
    if(oldVal != null) oldVal else newQueue.take
  }

  E poll() {
    E oldVal = oldPoll
    if(oldVal != null) oldVal else newQueue.poll
  }

  private E oldPoll() {
    // If you have more than one consumer thread, then use a temporary variable
    // for oldQueue - otherwise it might be set to null between the null check
    // and the call to poll
    if(oldQueue == null) null
    else {
      E oldVal = oldQueue.poll
      if(oldVal != null) oldVal
      else {
        oldQueue = null
        null
      }
    }
  }
}

Answer 2

对于您的问题：

1) 没有一个允许您手动更改队列大小，尽管 LinkedBlockingQueue 之类的东西会增长到您为其设置的最大值。

2 和 3）您可以使用文档中描述的第 3 个构造函数执行您所描述的操作（创建一个新的 ArrayBlockingQueue）：

ArrayBlockingQueue(int capacity, boolean fair, Collection c)

创建一个具有给定（固定）容量、指定访问策略和最初包含给定集合的元素的 ArrayBlockingQueue，按集合迭代器的遍历顺序添加。

这将为您提供您正在寻找的副本结构，并允许您设置新容量。调整大小：

// create the first queue
Queue smallQueue = new ArrayBlockingQueue(5);

// copy small queue over to big queue
Queue bigQueue = new ArrayBlockingQueue(10, false, smallQueue);

缩小尺寸（伪代码）：

Queue bigQueue = new ArrayBlockingQueue(10);
// start processing data with your producer / consumer.

// then...
Queue smallQueue = new ArrayBlockingQueue(1);
// 1) change producer to start doing puts into the smallQueue
// 2) let consumer continue consuming from the bigQueue until it is empty
// 3) change consumer to start polling from the smallQueue

在您切换消费者之前，您从第 1 步开始的 put 将被阻塞。