【问题标题】:How to block until a BlockingQueue is empty?如何阻塞直到 BlockingQueue 为空?
【发布时间】:2013-03-03 10:57:34
【问题描述】:

我正在寻找一种阻止 BlockingQueue 为空的方法。

我知道,在多线程环境中,只要有生产者将项目放入BlockingQueue,就可能会出现队列为空,几纳秒后满是项目的情况。

但是,如果只有一个生产者,那么它可能希望在停止将项目放入队列后等待(并阻塞)直到队列为空。

Java/伪代码:

// Producer code
BlockingQueue queue = new BlockingQueue();

while (having some tasks to do) {
    queue.put(task);
}

queue.waitUntilEmpty(); // <-- how to do this?

print("Done");

你有什么想法吗?

编辑:我知道包装 BlockingQueue 并使用额外的条件可以解决问题,我只是想问是否有一些预制的解决方案和/或更好的替代方案。

【问题讨论】:

  • 显然你可以调用peek,直到它返回null。阻止是什么使得这是一个不可接受的解决方案?
  • AFAIK 对您的编辑的回答是否定的。请注意,正如我在回答中所写的那样,您的用例非常奇特......确保您确实需要做您要求的事情。你没有说明为什么你需要这样的行为。
  • @JoãoFernandes:我现在不需要它,只是出于好奇。我喜欢阅读有关编程问题的意见。
  • 现在这是一个很好的论点 :) 让我们等待更多答案,看看是否有人知道我们不知道的事情。

标签: java concurrency blockingqueue


【解决方案1】:

使用wait()notify() 的简单解决方案:

// Producer:
synchronized(queue) {
    while (!queue.isEmpty())
        queue.wait(); //wait for the queue to become empty
    queue.put();
}

//Consumer:
synchronized(queue) {
    queue.get();
    if (queue.isEmpty())
        queue.notify(); // notify the producer
}

【讨论】:

  • 嗨@niculare,这段代码会导致死锁吗?两个代码 sn-ps 锁定同一个对象queue,防止生产者和消费者同时工作。因此,当生产者等待队列为空时,队列可能永远不会为空。
  • @liuyaodong 不,阅读Object.wait() 的Javadoc。它释放锁,直到另一个线程调用notify。它甚至明确表示“当前线程必须拥有这个对象的监视器”。
  • 这将如何工作?假设消费者首先运行并在 queue.get() 上被阻止,因为队列中还没有项目。现在生产者无法生产,因为它无法获得锁。
  • BlockingQueue 没有 get 方法。你是说poll 还是take
  • 我的意思是take,因为poll 在这种情况下不会有任何影响
【解决方案2】:

我知道您可能已经有一堆线程在主动轮询或排队,但我仍然觉得您的流程/设计不太正确。

队列变空并不意味着之前添加的任务已经完成,有些项目可能需要很长时间才能处理,因此检查是否为空并没有太大用处。

所以你应该做的是忘记BlockingQueue,你可以将它用作任何其他集合。将项目翻译成CollectionsCallable 并使用ExecutorService.invokeAll()

    Collection<Item> queue = ...
    Collection<Callable<Result>> tasks = new ArrayList<Callable<Result>>();

    for (Item item : queue) {
        tasks.add(new Callable<Result>() {

            @Override
            public Result call() throws Exception {
                // process the item ...

                return result;
            }
        });
    }

    // look at the results, add timeout for invokeAll if necessary
    List<Future<Result>> results = executorService.invokeAll(tasks);

    // done

这种方法可以让您完全控制生产者可以等待多长时间以及适当的异常处理。

【讨论】:

    【解决方案3】:

    这不是您想要做的,但使用 SynchronousQueue 会产生与您的 Java/伪代码非常相似的效果,即生产者阻塞,直到某个消费者检索到所有数据。

    唯一的区别是生产者在每个 put 上阻塞,直到消费者来检索数据,而不是最后一次。不确定这是否会对您的情况产生影响。如果生产者执行的任务有点昂贵,我希望它只会产生明显的不同。

    【讨论】:

      【解决方案4】:

      您的用例应该非常特殊,因为大多数情况下您只想在队列已满时阻塞生产者,而不是等到它为空。

      无论如何,这是可行的。我相信旋转直到isEmpty 返回 true 并没有那么低效,因为生产者将在本地旋转,即将访问自己的缓存,而不是撞到总线。但是,由于线程仍可调度,因此会消耗 CPU 时间。但是本地旋转绝对是更简单的方法。否则我会看到两个选项:

      1. 使用wait + notify 就像@niculare 建议的那样
      2. 不知何故,让第一个注意到队列为空的消费者以无锁方式通知生产者;这会更慢,但会优雅地降级“更多”

      【讨论】:

        【解决方案5】:

        @niculare 给出的答案似乎还可以,但请查看@hankduan 的评论。

        我在寻找类似问题的解决方案时看到了这个问题。
        最后,我或多或少地重写了LinkedBlockingQueue
        它具有其方法的子集,并且不实现 CollectionIterable
        它管理多个生产者和消费者。
        对我来说它有效。

        /**********************************************************************************************************************
         * Import specifications
         *********************************************************************************************************************/
        import java.util.Date;
        import java.util.concurrent.TimeUnit;
        import java.util.concurrent.locks.*;
        
        /**********************************************************************************************************************
         * This class implements a completely reentrant FIFO.
         *********************************************************************************************************************/
        public class BlockingFIFO<E>
        {
          /********************************************************************************************************************
           * The constructor creates an empty FIFO with a capacity of {@link Integer#MAX_VALUE}.
           *******************************************************************************************************************/
          public BlockingFIFO()
          {
            // -----------------
            // Initialize object
            // -----------------
            this(Integer.MAX_VALUE);
        
          } // constructor
        
          /********************************************************************************************************************
           * The constructor creates an empty FIFO with the specified capacity.
           *
           * @param capacity_ipar The maximum number of elements the FIFO may contain.
           *******************************************************************************************************************/
          public BlockingFIFO(int capacity_ipar)
          {
            // ---------------------
            // Initialize attributes
            // ---------------------
            lock_attr = new ReentrantLock();
            not_empty_attr = lock_attr.newCondition();
            not_full_attr = lock_attr.newCondition();
            head_attr = null;
            tail_attr = null;
            capacity_attr = capacity_ipar;
            size_attr = 0;
        
          } // constructor
        
          /********************************************************************************************************************
           * This method removes all of the elements from the FIFO.
           *
           * @return The number of elements in the FIFO before it was cleared.
           *******************************************************************************************************************/
          public int clear()
          {
            // -----------------
            // Initialize result
            // -----------------
            int result;
            result = 0;
        
            // ----------
            // Clear FIFO
            // ----------
            lock_attr.lock();
            try
            {
              result = size_attr;
              head_attr = null;
              tail_attr = null;
              size_attr = 0;
              not_full_attr.signalAll();
            }
            finally
            {
              lock_attr.unlock();
            }
        
            // ----
            // Done
            // ----
            return (result);
        
          } // clear
        
          /********************************************************************************************************************
           * This method returns the number of elements in the FIFO.
           *
           * @return The number of elements in the FIFO.
           *******************************************************************************************************************/
          public int size()
          {
            // -----------
            // Return size
            // -----------
            lock_attr.lock();
            try
            {
              return (size_attr);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // size
        
          /********************************************************************************************************************
           * This method returns the number of additional elements that the FIFO can ideally accept without blocking.
           *
           * @return The remaining capacity the FIFO.
           *******************************************************************************************************************/
          public int remainingCapacity()
          {
            // -------------------------
            // Return remaining capacity
            // -------------------------
            lock_attr.lock();
            try
            {
              return (capacity_attr - size_attr);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // remainingCapacity
        
          /********************************************************************************************************************
           * This method waits for the FIFO to become empty.
           *
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for the FIFO to become
           *                              empty.
           *******************************************************************************************************************/
          public void waitEmpty()
            throws InterruptedException
          {
            // -----------------------------
            // Wait for FIFO to become empty
            // -----------------------------
            lock_attr.lock();
            try
            {
              while (size_attr > 0)
                not_full_attr.await();
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // waitEmpty
        
          /********************************************************************************************************************
           * This method waits at most the specified time for the FIFO to become empty.
           * <br>It returns <code>true</code> if the FIFO is empty and <code>false</code> otherwise.
           *
           * @param  timeout_ipar         The maximum number of milliseconds to wait for the FIFO to become empty.
           * @return                      True if and only if the FIFO is empty.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for the FIFO to become
           *                              empty.
           *******************************************************************************************************************/
          public boolean waitEmpty(long timeout_ipar)
            throws InterruptedException
          {
            // ------------------
            // Determine deadline
            // ------------------
            Date deadline;
            deadline = new Date(System.currentTimeMillis() + timeout_ipar);
        
            // -----------------------------
            // Wait for FIFO to become empty
            // -----------------------------
            lock_attr.lock();
            try
            {
              while (size_attr > 0)
              {
                if (!not_full_attr.awaitUntil(deadline))
                  return (false);
              }
              return (true);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // waitEmpty
        
          /********************************************************************************************************************
           * This method waits at most the specified time for the FIFO to become empty.
           * <br>It returns <code>true</code> if the FIFO is empty and <code>false</code> otherwise.
           *
           * @param  timeout_ipar         The maximum time to wait for the FIFO to become empty.
           * @param  unit_ipar            The unit of the specified timeout.
           * @return                      True if and only if the FIFO is empty.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for the FIFO to become
           *                              empty.
           *******************************************************************************************************************/
          public boolean waitEmpty(long    timeout_ipar,
                                   TimeUnit unit_ipar)
            throws InterruptedException
          {
            // -----------------------------
            // Wait for FIFO to become empty
            // -----------------------------
            return (waitEmpty(unit_ipar.toMillis(timeout_ipar)));
        
          } // waitEmpty
        
          /********************************************************************************************************************
           * This method adds the specified element at the end of the FIFO if it is possible to do so immediately without
           * exceeding the queue's capacity.
           * <br>It returns <code>true</code> upon success and <code>false</code> if this queue is full.
           *
           * @param  element_ipar The element to add to the FIFO.
           * @return              True if and only if the element was added to the FIFO.
           *******************************************************************************************************************/
          public boolean offer(E element_ipar)
          {
            // ----------------------
            // Try to add the element
            // ----------------------
            lock_attr.lock();
            try
            {
              if (capacity_attr > size_attr)
              {
                push(element_ipar);
                return (true);
              }
              else
                return (false);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // offer
        
          /********************************************************************************************************************
           * This method adds the specified element at the end of the FIFO, waiting if necessary up to the specified wait time
           * for space to become available.
           * <br>It returns <code>true</code> upon success and <code>false</code> if this queue is full.
           *
           * @param  element_ipar         The element to add to the FIFO.
           * @param  timeout_ipar         The maximum number of milliseconds to wait for space to become available.
           * @return                      True if and only if the element was added to the FIFO.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for space to become
           *                              available.
           *******************************************************************************************************************/
          public boolean offer(E    element_ipar,
                               long timeout_ipar)
            throws InterruptedException
          {
            // ------------------
            // Determine deadline
            // ------------------
            Date deadline;
            deadline = new Date(System.currentTimeMillis() + timeout_ipar);
        
            // ----------------------
            // Try to add the element
            // ----------------------
            lock_attr.lock();
            try
            {
              while (size_attr == capacity_attr)
              {
                if (!not_full_attr.awaitUntil(deadline))
                  return (false);
              }
              push(element_ipar);
              return (true);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // offer
        
          /********************************************************************************************************************
           * This method adds the specified element at the end of the FIFO, waiting if necessary up to the specified wait time
           * for space to become available.
           * <br>It returns <code>true</code> upon success and <code>false</code> if this queue is full.
           *
           * @param  element_ipar         The element to add to the FIFO.
           * @param  timeout_ipar         The maximum time to wait for space to become available.
           * @param  unit_ipar            The unit of the specified timeout.
           * @return                      True if and only if the element was added to the FIFO.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for space to become
           *                              available.
           *******************************************************************************************************************/
          public boolean offer(E        element_ipar,
                               long     timeout_ipar,
                               TimeUnit unit_ipar)
            throws InterruptedException
          {
            // ----------------------------
            // Try to add specified element
            // ----------------------------
            return (offer(element_ipar, unit_ipar.toMillis(timeout_ipar)));
        
          } // offer
        
          /********************************************************************************************************************
           * This method adds the specified element at the end of the FIFO, waiting if necessary for space to become available.
           *
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for space to become
           *                              available.
           *******************************************************************************************************************/
          public void put(E element_ipar)
            throws InterruptedException
          {
            // ----------------------
            // Try to add the element
            // ----------------------
            lock_attr.lock();
            try
            {
              while (size_attr == capacity_attr)
                not_full_attr.await();
              push(element_ipar);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // put
        
          /********************************************************************************************************************
           * This method retrieves, but does not remove, the head of the FIFO, or returns <code>null</code> if the FIFO is
           * empty.
           *
           * @return The head of the FIFO, or <code>null</code> if the FIFO is empty.
           *******************************************************************************************************************/
          public E peek()
          {
            // --------------------
            // Return first element
            // --------------------
            lock_attr.lock();
            try
            {
              if (size_attr == 0)
                return (null);
              else
                return (head_attr.contents);
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // peek
        
          /********************************************************************************************************************
           * This method retrieves and removes the head of the FIFO, or returns <code>null</code> if the FIFO is
           * empty.
           *
           * @return The head of the FIFO, or <code>null</code> if the FIFO is empty.
           *******************************************************************************************************************/
          public E poll()
          {
            // --------------------
            // Return first element
            // --------------------
            lock_attr.lock();
            try
            {
              if (size_attr == 0)
                return (null);
              else
                return (pop());
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // poll
        
          /********************************************************************************************************************
           * This method retrieves and removes the head of the FIFO, waiting up to the specified wait time if necessary for an
           * element to become available.
           * <br>It returns <code>null</code> if the specified waiting time elapses before an element is available.
           *
           * @param  timeout_ipar         The maximum number of milliseconds to wait for an element to become available.
           * @return                      The head of the FIFO, or <code>null</code> if the specified waiting time elapses
           *                              before an element is available.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for an element to become
           *                              available.
           *******************************************************************************************************************/
          public E poll(long timeout_ipar)
            throws InterruptedException
          {
            // ------------------
            // Determine deadline
            // ------------------
            Date deadline;
            deadline = new Date(System.currentTimeMillis() + timeout_ipar);
        
            // --------------------
            // Return first element
            // --------------------
            lock_attr.lock();
            try
            {
              while (size_attr == 0)
              {
                if (!not_empty_attr.awaitUntil(deadline))
                  return (null);
              }
              return (pop());
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // poll
        
          /********************************************************************************************************************
           * This method retrieves and removes the head of the FIFO, waiting up to the specified wait time if necessary for an
           * element to become available.
           * <br>It returns <code>null</code> if the specified waiting time elapses before an element is available.
           *
           * @param  timeout_ipar         The maximum time to wait for an element to become available.
           * @param  unit_ipar            The unit of the specified timeout.
           * @return                      The head of the FIFO, or <code>null</code> if the specified waiting time elapses
           *                              before an element is available.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for an element to become
           *                              available.
           *******************************************************************************************************************/
          public E poll(long     timeout_ipar,
                        TimeUnit unit_ipar)
            throws InterruptedException
          {
            // ------------------------
            // Try to get first element
            // ------------------------
            return (poll(unit_ipar.toMillis(timeout_ipar)));
        
          } // poll
        
          /********************************************************************************************************************
           * This method retrieves and removes the head of the FIFO, waiting if necessary for an element to become available.
           *
           * @return                      The head of the FIFO.
           * @throws InterruptedException Thrown when the current thread got interrupted while waiting for space to become
           *                              available.
           *******************************************************************************************************************/
          public E take()
            throws InterruptedException
          {
            // ---------------------------
            // Try to return first element
            // ---------------------------
            lock_attr.lock();
            try
            {
              while (size_attr == 0)
                not_empty_attr.await();
              return (pop());
            }
            finally
            {
              lock_attr.unlock();
            }
        
          } // take
        
          /********************************************************************************************************************
           * This class implements as node within the FIFO.
           *******************************************************************************************************************/
          private class Node
          {
            E    contents;
            Node next;
        
          } // class Node
        
          /********************************************************************************************************************
           * This method adds the specified element to the end of the FIFO.
           * <br>It sends a signal to all threads waiting for the FIFO to contain something.
           * <br>The caller should have locked the object and have made sure the list is not full.
           *******************************************************************************************************************/
          private void push(E element_ipar)
          {
            // -----------
            // Create node
            // -----------
            Node node;
            node = new Node();
            node.contents = element_ipar;
            node.next = null;
        
            // --------------
            // Add to the end
            // --------------
            if (head_attr == null)
              head_attr = node;
            else
              tail_attr.next = node;
            tail_attr = node;
        
            // ----------------------
            // We got another element
            // ----------------------
            size_attr++;
            not_empty_attr.signalAll();
        
          } // push
        
          /********************************************************************************************************************
           * This method removes the first element from the FIFO and returns it.
           * <br>It sends a signal to all threads waiting for the FIFO to have space available.
           * <br>The caller should have locked the object and have made sure the list is not empty.
           *******************************************************************************************************************/
          private E pop()
          {
            // ------------
            // Isolate node
            // ------------
            Node node;
            node = head_attr;
            head_attr = node.next;
            if (head_attr == null)
              tail_attr = null;
        
            // --------------------------
            // We removed another element
            // --------------------------
            size_attr--;
            not_full_attr.signalAll();
        
            // ----
            // Done
            // ----
            return (node.contents);
        
          } // pop
        
          /********************************************************************************************************************
           * This attribute represents the lock on the FIFO.
           *******************************************************************************************************************/
          private Lock lock_attr;
        
          /********************************************************************************************************************
           * This attribute represents the condition of the FIFO not being empty.
           *******************************************************************************************************************/
          private Condition not_empty_attr;
        
          /********************************************************************************************************************
           * This attribute represents the condition of the FIFO not being full.
           *******************************************************************************************************************/
          private Condition not_full_attr;
        
          /********************************************************************************************************************
           * This attribute represents the first element of the FIFO.
           *******************************************************************************************************************/
          private Node head_attr;
        
          /********************************************************************************************************************
           * This attribute represents the last element of the FIFO.
           *******************************************************************************************************************/
          private Node tail_attr;
        
          /********************************************************************************************************************
           * This attribute represents the capacity of the FIFO.
           *******************************************************************************************************************/
          private int capacity_attr;
        
          /********************************************************************************************************************
           * This attribute represents the size of the FIFO.
           *******************************************************************************************************************/
          private int size_attr;
        
        } // class BlockingFIFO
        

        【讨论】:

          【解决方案6】:

          根据documentation, BlockingQueue 可以通过以下方法在“删除”时被阻止:take()poll(time, unit) 而不是 poll()

          【讨论】:

            猜你喜欢
            • 2012-01-05
            • 2014-02-08
            • 1970-01-01
            • 2010-10-23
            • 2017-04-16
            • 1970-01-01
            • 1970-01-01
            • 2014-08-26
            • 1970-01-01
            相关资源
            最近更新 更多