Compréhension approfondie du blocage des conteneurs de file d'attente dans la programmation de threads Java_Connaissances de base

1. Qu'est-ce qu'une file d'attente bloquante ?

BlockingQueue est une file d'attente qui prend en charge deux opérations supplémentaires. Ces deux opérations supplémentaires sont : lorsque la file d'attente est vide, le thread récupérant l'élément attend que la file d'attente devienne non vide. Lorsque la file d'attente est pleine, le thread stockant l'élément attend que la file d'attente soit disponible. Les files d'attente de blocage sont souvent utilisées dans les scénarios de producteur et de consommateur. Le producteur est le thread qui ajoute des éléments à la file d'attente et le consommateur est le thread qui extrait les éléments de la file d'attente. La file d'attente de blocage est un conteneur dans lequel le producteur stocke les éléments et le consommateur ne récupère que les éléments du conteneur.

La file d'attente de blocage propose quatre méthodes de traitement :

Lancement d'une exception : lorsque la file d'attente de blocage est pleine, l'insertion d'éléments dans la file d'attente lèvera une exception IllegalStateException ("Queue full"). Lorsque la file d'attente est vide, NoSuchElementException sera levée lors de l'obtention d'éléments de la file d'attente.
Renvoie une valeur spéciale : la méthode d'insertion indiquera si elle réussit ou non, et retournera vrai en cas de succès. La méthode de suppression consiste à retirer un élément de la file d'attente, s'il n'y a pas d'élément, renvoyer null
Toujours bloquant : lorsque la file d'attente de blocage est pleine, si le thread producteur place des éléments dans la file d'attente, la file d'attente bloquera le thread producteur jusqu'à ce qu'il récupère les données ou se termine en réponse à une interruption. Lorsque la file d'attente est vide et que le thread consommateur tente de récupérer des éléments de la file d'attente, la file d'attente bloque également le thread consommateur jusqu'à ce que la file d'attente soit disponible.
Sortie du délai d'attente : lorsque la file d'attente de blocage est pleine, la file d'attente bloquera le thread producteur pendant un certain temps. Si elle dépasse un certain temps, le thread producteur se fermera.
2. File d'attente bloquante en Java

JDK7 fournit 7 files d'attente de blocage. Ils sont

1. ArrayBlockingQueue : une file d'attente de blocage limitée composée d'une structure de tableau.
2. LinkedBlockingQueue : une file d'attente de blocage limitée composée d'une structure de liste chaînée.
3. PriorityBlockingQueue : une file d'attente de blocage illimitée qui prend en charge le tri prioritaire.
4. DelayQueue : une file d'attente de blocage illimitée implémentée à l'aide d'une file d'attente prioritaire.
5. SynchronousQueue : Une file d'attente de blocage qui ne stocke pas d'éléments.
6. LinkedTransferQueue : une file d'attente de blocage illimitée composée d'une structure de liste chaînée.
7. LinkedBlockingDeque : une file d'attente de blocage bidirectionnelle composée d'une structure de liste chaînée.

ArrayBlockingQueue est une file d'attente de blocage limitée implémentée à l'aide d'un tableau. Cette file d'attente trie les éléments selon le principe du premier entré, premier sorti (FIFO). Par défaut, les visiteurs n'ont pas la garantie d'un accès équitable à la file d'attente. La file d'attente dite d'accès équitable fait référence à tous les threads producteurs ou consommateurs bloqués. Lorsque la file d'attente est disponible, la file d'attente est accessible dans l'ordre de blocage, c'est-à-dire. le thread producteur bloqué en premier, vous pouvez d'abord insérer des éléments dans la file d'attente, et le thread consommateur qui bloque en premier peut d'abord obtenir des éléments de la file d'attente. Habituellement, le débit est réduit pour garantir l’équité. Nous pouvons créer une file d'attente de blocage équitable en utilisant le code suivant :

ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);

L'équité des visiteurs est obtenue grâce aux serrures réentrantes, le code est le suivant :

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 est une file d'attente de blocage limitée implémentée à l'aide d'une liste chaînée. La longueur par défaut et maximale de cette file d'attente est Integer.MAX_VALUE. Cette file d'attente trie les éléments selon le principe du premier entré, premier sorti.

PriorityBlockingQueue est une file d'attente illimitée qui prend en charge la priorité. Par défaut, les éléments sont disposés dans un ordre naturel et les règles de tri des éléments peuvent également être spécifiées via le comparateur. Les éléments sont triés par ordre croissant.

DelayQueue est une file d'attente de blocage illimitée qui prend en charge l'acquisition retardée d'éléments. Les files d'attente sont implémentées à l'aide de PriorityQueue. Les éléments de la file d'attente doivent implémenter l'interface Delayed. Lors de la création de l'élément, vous pouvez spécifier le temps qu'il faudra pour extraire l'élément actuel de la file d'attente. Les éléments ne peuvent être extraits de la file d'attente qu'à l'expiration du délai. Nous pouvons utiliser DelayQueue dans les scénarios d'application suivants :

Conception du système de cache : vous pouvez utiliser DelayQueue pour enregistrer la période de validité des éléments du cache et utiliser un thread pour interroger le DelayQueue dans une boucle. Une fois que l'élément peut être obtenu à partir du DelayQueue, cela signifie que la période de validité du cache est terminée. expiré.
Tâches planifiées. Utilisez DelayQueue pour enregistrer les tâches et le temps d'exécution qui seront exécutés ce jour-là. Une fois la tâche obtenue de DelayQueue, elle sera exécutée. Par exemple, TimerQueue est implémenté à l'aide de DelayQueue.
Retardé dans la file d'attente doit implémenter compareTo pour spécifier l'ordre des éléments. Par exemple, placez celui avec le retard le plus long en fin de file d’attente. Le code d'implémentation est le suivant :

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. Comment implémenter l'interface différée

Nous pouvons faire référence à la classe ScheduledFutureTask dans ScheduledThreadPoolExecutor. Cette classe implémente l'interface Delayed. Premièrement : lorsque l'objet est créé, utilisez le temps pour enregistrer le moment où l'objet précédent peut être utilisé. Le code est le suivant :

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

Utilisez ensuite getDelay pour demander combien de temps l'élément actuel doit être retardé. Le code est le suivant :

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()和非阻塞队列实现生产者-消费者的思路类似，只不过它把这些工作一起集成到了阻塞队列中实现。