Java阻塞隊列

阻塞隊列通過添加兩組方法來擴展隊列：

• 一組方法無限期地阻塞

• 另一組方法允許您指定要阻止的時間段。

BlockingQueue 接口的實例表示阻塞隊列。 BlockingQueue 接口繼承自 Queue 接口。

put()和 offer()方法在阻塞隊列的尾部添加一個元素。如果阻塞隊列已滿，則put()方法將無限期阻塞，直到隊列中的空間可用。offer()方法允許您指定等待空間可用的時間段。 如果指定的元素添加成功，則返回true; 否則為假。

take()和poll()方法檢索和刪除阻塞隊列的頭。如果阻塞隊列為空，take()方法將無限期阻塞。poll()方法允許您指定在阻塞隊列為空時要等待的時間段; 如果在元素可用之前過去了指定的時間，則返回null。

來自 BlockingQueue 中 Queue 接口的方法就像使用 Queue 。

BlockingQueue 被設計為線程安全的并且可以使用在生產者/消費者的情況下。

阻塞隊列不允許空元素和可以是有界的或無界的。

BlockingQueue 中的 remainingCapacity()返回可以添加到阻止隊列中而不阻塞的元素數。

BlockingQueue 可以控制多個線程被阻塞時的公平性。 如果阻塞隊列是公平的，它可以選擇最長的等待線程來執(zhí)行操作。如果阻塞隊列不公平，則不指定選擇的順序。

BlockingQueue 接口及其所有實現類都在 java.util.concurrent 包中。 以下是 BlockingQueue 接口的實現類:

由數組支持的 ArrayBlockingQueue 是 BlockingQueue 的有界實現類。 我們可以在其構造函數中指定阻塞隊列的公平性。 默認情況下，它不公平。

LinkedBlockingQueue 可以用作有界或無界阻塞隊列。 它不允許為阻塞隊列指定公平規(guī)則。

PriorityBlockingQueue 是 BlockingQueue 的無界實現類。 它的工作方式與 PriortyQueue 相同，用于排序阻塞隊列中的元素，并將阻塞特性添加到 PriorityQueue 中。

SynchronousQueue 實現 BlockingQueue ，沒有任何容量。 put操作等待take操作以獲取元素。 它可以在兩個線程之間進行握手，并在兩個線程之間交換對象。 它的isEmpty()方法總是返回true。

DelayQueue是BlockingQueue的無界實現類。它保持一個元素，直到該元素經過指定的延遲。 如果有超過一個元素的延遲已經過去，那么其延遲最早傳遞的元素將被放置在隊列的頭部。

例子

以下代碼顯示了如何在生產者/消費者應用程序中使用阻塞隊列。

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

class BQProducer extends Thread {
  private final BlockingQueue queue;
  private final String name;
  public BQProducer(BlockingQueue queue, String name) {
    this.queue = queue;
    this.name = name;
  }
  @Override
  public void run() {
    while (true) {
      try {
        this.queue.put(UUID.randomUUID().toString());
        Thread.sleep(4000);
      }
      catch (InterruptedException e) {
        e.printStackTrace();
        break;
      }
    }
  }
}
class BQConsumer extends Thread {
  private final BlockingQueue queue;
  private final String name;
  public BQConsumer(BlockingQueue queue, String name) {
    this.queue = queue;
    this.name = name;
  }

  @Override
  public void run() {
    while (true) {
      try {
        String str = this.queue.take();
        System.out.println(name + "  took: " + str);
        Thread.sleep(3000);
      } catch (InterruptedException e) {
        e.printStackTrace();
        break;
      }
    }
  }
}

public class Main {
  public static void main(String[] args) {
    int capacity = 5;
    boolean fair = true;
    BlockingQueue queue = new ArrayBlockingQueue<>(capacity, fair);

    new BQProducer(queue, "Producer1").start();
    new BQProducer(queue, "Producer2").start();
    new BQProducer(queue, "Producer3").start();
    new BQConsumer(queue, "Consumer1").start();
    new BQConsumer(queue, "Consumer2").start();
  }
}

上面的代碼生成以下結果。