Vertiefte Kenntnisse zum Blockieren von Warteschlangencontainern in der Java-Thread-Programmierung_Grundkenntnisse

1. Was ist eine Blockierungswarteschlange?

BlockingQueue ist eine Warteschlange, die zwei zusätzliche Vorgänge unterstützt. Diese beiden zusätzlichen Vorgänge sind: Wenn die Warteschlange leer ist, wartet der Thread, der das Element abruft, darauf, dass die Warteschlange nicht mehr leer ist. Wenn die Warteschlange voll ist, wartet der Thread, der das Element speichert, darauf, dass die Warteschlange verfügbar wird. Blockierende Warteschlangen werden häufig in Produzenten- und Verbraucherszenarien verwendet. Der Produzent ist der Thread, der Elemente zur Warteschlange hinzufügt, und der Verbraucher ist der Thread, der Elemente aus der Warteschlange übernimmt. Die Blockierungswarteschlange ist ein Container, in dem der Produzent Elemente speichert und der Verbraucher nur Elemente aus dem Container entnimmt.

Die Blockierungswarteschlange bietet vier Verarbeitungsmethoden:

Auslösen einer Ausnahme: Wenn die blockierende Warteschlange voll ist, wird beim Einfügen von Elementen in die Warteschlange eine IllegalStateException („Warteschlange voll“)-Ausnahme ausgelöst. Wenn die Warteschlange leer ist, wird beim Abrufen von Elementen aus der Warteschlange eine NoSuchElementException ausgelöst.
Sonderwert zurückgeben: Die Einfügemethode gibt zurück, ob sie erfolgreich ist oder nicht, und gibt bei Erfolg true zurück. Die Entfernungsmethode besteht darin, ein Element aus der Warteschlange zu entfernen. Wenn kein Element vorhanden ist, wird null zurückgegeben
Immer blockierend: Wenn die Blockierungswarteschlange voll ist und der Produzenten-Thread Elemente in die Warteschlange einfügt, blockiert die Warteschlange den Produzenten-Thread, bis er die Daten erhält oder als Reaktion auf einen Interrupt beendet wird. Wenn die Warteschlange leer ist und der Verbraucherthread versucht, Elemente aus der Warteschlange zu übernehmen, blockiert die Warteschlange auch den Verbraucherthread, bis die Warteschlange verfügbar ist.
Timeout-Ausgang: Wenn die Blockierungswarteschlange voll ist, blockiert die Warteschlange den Produzenten-Thread für einen bestimmten Zeitraum. Wenn sie eine bestimmte Zeit überschreitet, wird der Produzenten-Thread beendet.
2. Blockierungswarteschlange in Java

JDK7 bietet 7 Blockierungswarteschlangen. Sie sind

1. ArrayBlockingQueue: Eine begrenzte Blockierungswarteschlange, die aus einer Array-Struktur besteht.
2. LinkedBlockingQueue: Eine begrenzte Blockierungswarteschlange, die aus einer verknüpften Listenstruktur besteht.
3. PriorityBlockingQueue: Eine unbegrenzte Blockierungswarteschlange, die die Prioritätssortierung unterstützt.
4. DelayQueue: Eine unbegrenzte Blockierungswarteschlange, die mithilfe einer Prioritätswarteschlange implementiert wird.
5. SynchronousQueue: Eine blockierende Warteschlange, die keine Elemente speichert.
6. LinkedTransferQueue: eine unbegrenzte Blockierungswarteschlange, die aus einer verknüpften Listenstruktur besteht.
7. LinkedBlockingDeque: Eine bidirektionale Blockierungswarteschlange, die aus einer verknüpften Listenstruktur besteht.

ArrayBlockingQueue ist eine begrenzte Blockierungswarteschlange, die mithilfe eines Arrays implementiert wird. Diese Warteschlange sortiert Elemente nach dem First-In-First-Out-Prinzip (FIFO). Standardmäßig wird Besuchern kein fairer Zugriff auf die Warteschlange garantiert. Die sogenannte Fair-Access-Warteschlange bezieht sich auf alle blockierten Produzenten-Threads oder Verbraucher-Threads. Wenn die Warteschlange verfügbar ist, kann auf die Warteschlange in der Reihenfolge der Blockierung zugegriffen werden. Der Producer-Thread wird zuerst blockiert. Sie können zuerst Elemente in die Warteschlange einfügen, und der Consumer-Thread, der zuerst blockiert, kann zuerst Elemente aus der Warteschlange abrufen. Normalerweise wird der Durchsatz reduziert, um Fairness zu gewährleisten. Mit dem folgenden Code können wir eine faire Blockierungswarteschlange erstellen:

ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);

Besuchergerechtigkeit wird durch Wiedereintrittssperren erreicht, der Code lautet wie folgt:

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
      throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull = lock.newCondition();
}

LinkedBlockingQueue ist eine begrenzte Blockierungswarteschlange, die mithilfe einer verknüpften Liste implementiert wird. Die Standard- und Maximallänge dieser Warteschlange ist Integer.MAX_VALUE. Diese Warteschlange sortiert Elemente nach dem First-In-First-Out-Prinzip.

PriorityBlockingQueue ist eine unbegrenzte Warteschlange, die Priorität unterstützt. Standardmäßig sind die Elemente in natürlicher Reihenfolge angeordnet, und die Sortierregeln der Elemente können auch über den Komparator angegeben werden. Die Elemente werden in aufsteigender Reihenfolge sortiert.

DelayQueue ist eine unbegrenzte Blockierungswarteschlange, die die verzögerte Erfassung von Elementen unterstützt. Warteschlangen werden mit PriorityQueue implementiert. Die Elemente in der Warteschlange müssen die verzögerte Schnittstelle implementieren. Beim Erstellen des Elements können Sie angeben, wie lange es dauern soll, das aktuelle Element aus der Warteschlange zu erhalten. Elemente können erst aus der Warteschlange abgerufen werden, wenn die Verzögerung abgelaufen ist. Wir können DelayQueue in folgenden Anwendungsszenarien verwenden:

Cache-Systemdesign: Sie können DelayQueue verwenden, um die Gültigkeitsdauer von Cache-Elementen zu speichern, und einen Thread verwenden, um die DelayQueue in einer Schleife abzufragen. Sobald das Element aus der DelayQueue abgerufen werden kann, bedeutet dies, dass die Cache-Gültigkeitsdauer abgelaufen ist Abgelaufen.
Geplante Aufgaben. Verwenden Sie DelayQueue, um die Aufgaben und die Ausführungszeit zu speichern, die an dem Tag ausgeführt werden. Sobald die Aufgabe von DelayQueue abgerufen wird, wird sie beispielsweise mit DelayQueue ausgeführt.
Verzögerungen in der Warteschlange müssen CompareTo implementieren, um die Reihenfolge der Elemente anzugeben. Stellen Sie beispielsweise denjenigen mit der längsten Verzögerung an das Ende der Warteschlange. Der Implementierungscode lautet wie folgt:

public int compareTo(Delayed other) {
      if (other == this) // compare zero ONLY if same object
        return 0;
      if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask x = (ScheduledFutureTask)other;
        long diff = time - x.time;
        if (diff < 0)
          return -1;
        else if (diff > 0)
          return 1;
  else if (sequenceNumber < x.sequenceNumber)
          return -1;
        else
          return 1;
      }
      long d = (getDelay(TimeUnit.NANOSECONDS) -
           other.getDelay(TimeUnit.NANOSECONDS));
      return (d == 0) &#63; 0 : ((d < 0) &#63; -1 : 1);
    }

3. So implementieren Sie die verzögerte Schnittstelle

Wir können auf die ScheduledFutureTask-Klasse in ScheduledThreadPoolExecutor verweisen. Diese Klasse implementiert die Delayed-Schnittstelle. Erstens: Verwenden Sie beim Erstellen des Objekts die Zeit, um aufzuzeichnen, wann das vorherige Objekt verwendet werden kann. Der Code lautet wie folgt:

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
      super(r, result);
      this.time = ns;
      this.period = period;
      this.sequenceNumber = sequencer.getAndIncrement();
}

Dann verwenden Sie getDelay, um abzufragen, wie lange das aktuelle Element verzögert werden muss. Der Code lautet wie folgt:

public long getDelay(TimeUnit unit) {
      return unit.convert(time - now(), TimeUnit.NANOSECONDS);
    }

通过构造函数可以看出延迟时间参数ns的单位是纳秒，自己设计的时候最好使用纳秒，因为getDelay时可以指定任意单位，一旦以纳秒作为单位，而延时的时间又精确不到纳秒就麻烦了。使用时请注意当time小于当前时间时，getDelay会返回负数。

4.如何实现延时队列

延时队列的实现很简单，当消费者从队列里获取元素时，如果元素没有达到延时时间，就阻塞当前线程。

long delay = first.getDelay(TimeUnit.NANOSECONDS);
          if (delay <= 0)
            return q.poll();
          else if (leader != null)
            available.await();

SynchronousQueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素。SynchronousQueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合于传递性场景,比如在一个线程中使用的数据，传递给另外一个线程使用，SynchronousQueue的吞吐量高于LinkedBlockingQueue 和 ArrayBlockingQueue。

LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。

transfer方法。如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer方法可以把生产者传入的元素立刻transfer（传输）给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。transfer方法的关键代码如下：

Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

第一行代码是试图把存放当前元素的s节点作为tail节点。第二行代码是让CPU自旋等待消费者消费元素。因为自旋会消耗CPU，所以自旋一定的次数后使用Thread.yield()方法来暂停当前正在执行的线程，并执行其他线程。

tryTransfer方法。则是用来试探下生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回。而transfer方法是必须等到消费者消费了才返回。

对于带有时间限制的tryTransfer(E e, long timeout, TimeUnit unit)方法，则是试图把生产者传入的元素直接传给消费者，但是如果没有消费者消费该元素则等待指定的时间再返回，如果超时还没消费元素，则返回false，如果在超时时间内消费了元素，则返回true。

LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的你可以从队列的两端插入和移出元素。双端队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。相比其他的阻塞队列，LinkedBlockingDeque多了addFirst，addLast，offerFirst，offerLast，peekFirst，peekLast等方法，以First单词结尾的方法，表示插入，获取（peek）或移除双端队列的第一个元素。以Last单词结尾的方法，表示插入，获取或移除双端队列的最后一个元素。另外插入方法add等同于addLast，移除方法remove等效于removeFirst。但是take方法却等同于takeFirst，不知道是不是Jdk的bug，使用时还是用带有First和Last后缀的方法更清楚。

在初始化LinkedBlockingDeque时可以设置容量防止其过渡膨胀。另外双向阻塞队列可以运用在“工作窃取”模式中。

5.阻塞队列的实现原理
本文以ArrayBlockingQueue为例，其他阻塞队列实现原理可能和ArrayBlockingQueue有一些差别，但是大体思路应该类似，有兴趣的朋友可自行查看其他阻塞队列的实现源码。

　　首先看一下ArrayBlockingQueue类中的几个成员变量：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
 
private static final long serialVersionUID = -817911632652898426L;
 
/** The queued items */
private final E[] items;
/** items index for next take, poll or remove */
private int takeIndex;
/** items index for next put, offer, or add. */
private int putIndex;
/** Number of items in the queue */
private int count;
 
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
 
/** Main lock guarding all access */
private final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
}

 　　可以看出，ArrayBlockingQueue中用来存储元素的实际上是一个数组，takeIndex和putIndex分别表示队首元素和队尾元素的下标，count表示队列中元素的个数。

　　lock是一个可重入锁，notEmpty和notFull是等待条件。

　　下面看一下ArrayBlockingQueue的构造器，构造器有三个重载版本：

public ArrayBlockingQueue(int capacity) {
}
public ArrayBlockingQueue(int capacity, boolean fair) {
 
}
public ArrayBlockingQueue(int capacity, boolean fair,
             Collection<&#63; extends E> c) {
}

 　　第一个构造器只有一个参数用来指定容量，第二个构造器可以指定容量和公平性，第三个构造器可以指定容量、公平性以及用另外一个集合进行初始化。

　　然后看它的两个关键方法的实现：put()和take()：

public void put(E e) throws InterruptedException {
  if (e == null) throw new NullPointerException();
  final E[] items = this.items;
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
    try {
      while (count == items.length)
        notFull.await();
    } catch (InterruptedException ie) {
      notFull.signal(); // propagate to non-interrupted thread
      throw ie;
    }
    insert(e);
  } finally {
    lock.unlock();
  }
}

 　　从put方法的实现可以看出，它先获取了锁，并且获取的是可中断锁，然后判断当前元素个数是否等于数组的长度，如果相等，则调用notFull.await()进行等待，如果捕获到中断异常，则唤醒线程并抛出异常。

　　当被其他线程唤醒时，通过insert(e)方法插入元素，最后解锁。

　　我们看一下insert方法的实现：

private void insert(E x) {
  items[putIndex] = x;
  putIndex = inc(putIndex);
  ++count;
  notEmpty.signal();
}

 　　它是一个private方法，插入成功后，通过notEmpty唤醒正在等待取元素的线程。

　　下面是take()方法的实现：

public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
    try {
      while (count == 0)
        notEmpty.await();
    } catch (InterruptedException ie) {
      notEmpty.signal(); // propagate to non-interrupted thread
      throw ie;
    }
    E x = extract();
    return x;
  } finally {
    lock.unlock();
  }
}

 　　跟put方法实现很类似，只不过put方法等待的是notFull信号，而take方法等待的是notEmpty信号。在take方法中，如果可以取元素，则通过extract方法取得元素，下面是extract方法的实现：

private E extract() {
  final E[] items = this.items;
  E x = items[takeIndex];
  items[takeIndex] = null;
  takeIndex = inc(takeIndex);
  --count;
  notFull.signal();
  return x;
}

 　　跟insert方法也很类似。

　　其实从这里大家应该明白了阻塞队列的实现原理，事实它和我们用Object.wait()、Object.notify()和非阻塞队列实现生产者-消费者的思路类似，只不过它把这些工作一起集成到了阻塞队列中实现。