Pemahaman mendalam tentang menyekat bekas baris gilir dalam pengaturcaraan benang Java_Pengetahuan asas

1. Apakah itu baris gilir menyekat?

BlockingQueue ialah baris gilir yang menyokong dua operasi tambahan. Dua operasi tambahan ini ialah: apabila baris gilir kosong, benang yang mendapat elemen menunggu baris gilir menjadi tidak kosong. Apabila baris gilir penuh, benang yang menyimpan elemen menunggu baris gilir tersedia. Menyekat baris gilir sering digunakan dalam senario pengeluar dan pengguna Pengeluar ialah utas yang menambah elemen pada baris gilir, dan pengguna ialah utas yang mengambil elemen daripada baris gilir. Barisan yang menyekat ialah bekas tempat pengeluar menyimpan elemen, dan pengguna hanya mengambil elemen daripada bekas.

Baris gilir menyekat menyediakan empat kaedah pemprosesan:

Melemparkan pengecualian: Apabila baris gilir menyekat penuh, memasukkan elemen ke dalam baris gilir akan membuang pengecualian IllegalStateException ("Baris gilir penuh". Apabila baris gilir kosong, NoSuchElementException akan dilemparkan apabila mendapat elemen daripada baris gilir.
Kembalikan nilai istimewa: Kaedah sisipan akan kembali sama ada ia berjaya atau tidak, dan kembali benar jika berjaya. Kaedah penyingkiran adalah untuk mengeluarkan elemen dari baris gilir, jika tiada elemen, kembalikan null
Sentiasa menyekat: Apabila baris gilir menyekat penuh, jika urutan pengeluar meletakkan elemen ke dalam baris gilir, baris gilir akan menyekat urutan pengeluar sehingga ia mendapat data atau keluar sebagai tindak balas kepada gangguan. Apabila baris gilir kosong dan urutan pengguna cuba untuk mengambil elemen daripada baris gilir, baris gilir juga akan menyekat urutan pengguna sehingga baris gilir tersedia.
Keluar masa tamat: Apabila baris gilir menyekat penuh, baris gilir akan menyekat benang pengeluar untuk tempoh masa Jika melebihi masa tertentu, urutan pengeluar akan keluar.
2. Menyekat baris gilir dalam Java

JDK7 menyediakan 7 barisan menyekat. Mereka adalah

1. ArrayBlockingQueue: Baris sekatan terhad yang terdiri daripada struktur tatasusunan.
2. LinkedBlockingQueue: Baris sekatan terhad yang terdiri daripada struktur senarai terpaut.
3. PriorityBlockingQueue: Baris sekatan tanpa sempadan yang menyokong pengisihan keutamaan.
4. DelayQueue: Barisan gilir sekatan tanpa sempadan dilaksanakan menggunakan baris gilir keutamaan.
5. SynchronousQueue: Barisan gilir menyekat yang tidak menyimpan elemen.
6. LinkedTransferQueue: baris gilir sekatan tanpa sempadan yang terdiri daripada struktur senarai terpaut.
7. LinkedBlockingDeque: Barisan gilir sekatan dua hala yang terdiri daripada struktur senarai terpaut.

ArrayBlockingQueue ialah baris gilir sekatan terhad yang dilaksanakan menggunakan tatasusunan. Barisan gilir ini mengisih elemen berdasarkan asas masuk dahulu keluar (FIFO). Secara lalai, pelawat tidak dijamin akses adil kepada baris gilir yang dipanggil baris gilir akses adil merujuk kepada semua benang pengeluar atau benang pengguna yang disekat Apabila baris gilir tersedia, baris gilir boleh diakses mengikut urutan penyekatan, iaitu, benang pengeluar disekat dahulu , anda boleh memasukkan elemen ke dalam baris gilir dahulu, dan benang pengguna yang menyekat dahulu boleh mendapatkan elemen daripada baris gilir dahulu. Biasanya, daya pengeluaran dikurangkan untuk memastikan keadilan. Kita boleh membuat baris gilir menyekat yang adil menggunakan kod berikut:

ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);

Kesaksamaan pengunjung dicapai menggunakan kunci reentrant, kodnya adalah seperti berikut:

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 ialah baris gilir sekatan terhad yang dilaksanakan menggunakan senarai terpaut. Panjang lalai dan maksimum baris gilir ini ialah Integer.MAX_VALUE. Baris gilir ini mengisih elemen mengikut dasar masuk dahulu, keluar dahulu.

PriorityBlockingQueue ialah baris gilir tanpa had yang menyokong keutamaan. Secara lalai, unsur-unsur disusun dalam susunan semula jadi, dan peraturan pengisihan unsur-unsur juga boleh ditentukan melalui pembanding. Elemen disusun mengikut tertib menaik.

DelayQueue ialah baris gilir sekatan tanpa sempadan yang menyokong pemerolehan elemen yang tertangguh. Baris gilir dilaksanakan menggunakan PriorityQueue. Elemen dalam baris gilir mesti melaksanakan antara muka Tertunda Apabila mencipta elemen, anda boleh menentukan tempoh masa yang diperlukan untuk mendapatkan elemen semasa daripada baris gilir. Elemen hanya boleh diambil daripada baris gilir apabila kelewatan tamat tempoh. Kita boleh menggunakan DelayQueue dalam senario aplikasi berikut:

Reka bentuk sistem cache: Anda boleh menggunakan DelayQueue untuk menyimpan tempoh sah elemen cache dan menggunakan benang untuk menanyakan DelayQueue dalam gelung Sebaik sahaja elemen itu boleh diperolehi daripada DelayQueue, ini bermakna tempoh sah cache telah tamat tempoh.
Tugas yang dijadualkan. Gunakan DelayQueue untuk menyimpan tugasan dan masa pelaksanaan yang akan dilaksanakan pada hari tersebut Setelah tugasan diperoleh daripada DelayQueue, ia akan dilaksanakan Sebagai contoh, TimerQueue dilaksanakan menggunakan DelayQueue.
Kelewatan dalam baris gilir mesti melaksanakan compareTo untuk menentukan susunan elemen. Sebagai contoh, letakkan yang mempunyai kelewatan paling lama di penghujung baris gilir. Kod pelaksanaan adalah seperti berikut:

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. Bagaimana untuk melaksanakan antara muka Tertunda

Kita boleh merujuk kepada kelas ScheduledFutureTask dalam ScheduledThreadPoolExecutor. Kelas ini melaksanakan antara muka Tertunda. Pertama: apabila objek dicipta, gunakan masa untuk merekodkan apabila objek sebelumnya boleh digunakan Kodnya adalah seperti berikut:

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

Kemudian gunakan getDelay untuk menanyakan berapa lama elemen semasa perlu ditangguhkan Kodnya adalah seperti berikut:

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