Java并发 - BlockingQueue接口

java.util.concurrent.BlockingQueue 接口是 Queue 接口的子接口，此外还支持诸如等待队列变为非空后再检索元素，以及等待队列中有空间后再存储元素之类的操作。

BlockingQueue方法

序号	方法及描述
1	boolean add(E e) 如果可以立即插入指定的元素而不会违反容量限制，则将该元素插入此队列中，成功时返回 true，如果没有可用空间则抛出 IllegalStateException。
2	boolean contains(Object o) 如果此队列包含指定的元素，则返回 true。
3	int drainTo(Collection<? super E> c) 从此队列中移除所有可用元素并将它们添加到给定的集合中。
4	int drainTo(Collection<? super E> c, int maxElements) 从此队列中移除最多给定数量的可用元素并将它们添加到给定的集合中。
5	boolean offer(E e) 如果可以立即插入指定的元素而不会违反容量限制，则将该元素插入此队列中，成功时返回 true，如果没有可用空间则返回 false。
6	boolean offer(E e, long timeout, TimeUnit unit) 将指定的元素插入此队列，如果必要，则等待直到指定等待时间，直到有空间可用。
7	E poll(long timeout, TimeUnit unit) 检索并移除此队列的头，如果必要，则等待直到指定等待时间，直到有元素可用。
8	void put(E e) 将指定的元素插入此队列，如有必要，则等待直到有空间可用。
9	int remainingCapacity() 返回此队列理想情况下（在没有内存或资源限制的情况下）可以在不阻塞的情况下接受的额外元素数量，如果没有固有限制，则返回 Integer.MAX_VALUE。
10	boolean remove(Object o) 从此队列中移除指定的元素的一个实例（如果存在）。
11	E take() 检索并移除此队列的头，如有必要，则等待直到有元素可用。

示例

下面的 TestThread 程序演示了在基于线程的环境中使用 BlockingQueue 接口。

在线演示

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class TestThread {

   public static void main(final String[] arguments) throws InterruptedException {
      BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(10);

      Producer producer = new Producer(queue);
      Consumer consumer = new Consumer(queue);

      new Thread(producer).start();
      new Thread(consumer).start();

      Thread.sleep(4000);
   }  


   static class Producer implements Runnable {
      private BlockingQueue<Integer> queue;

      public Producer(BlockingQueue queue) {
         this.queue = queue;
      }

      @Override
      public void run() {
         Random random = new Random();

         try {
            int result = random.nextInt(100);
            Thread.sleep(1000);
            queue.put(result);
            System.out.println("Added: " + result);
            
            result = random.nextInt(100);
            Thread.sleep(1000);
            queue.put(result);
            System.out.println("Added: " + result);
            
            result = random.nextInt(100);
            Thread.sleep(1000);
            queue.put(result);
            System.out.println("Added: " + result);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }	   
   }

   static class Consumer implements Runnable {
      private BlockingQueue<Integer> queue;

      public Consumer(BlockingQueue queue) {
         this.queue = queue;
      }
      
      @Override
      public void run() {
         
         try {
            System.out.println("Removed: " + queue.take());
            System.out.println("Removed: " + queue.take());
            System.out.println("Removed: " + queue.take());
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }
   }
}

这将产生以下结果。

输出

Added: 52
Removed: 52
Added: 70
Removed: 70
Added: 27
Removed: 27

打印页面