午夜手机在线,久久国产精品视频,欧美寡妇偷汉性猛交

大家好，我是小黑，一個(gè)在互聯(lián)網(wǎng)茍且偷生的農(nóng)民工。

隊(duì)列

學(xué)過(guò)數(shù)據(jù)結(jié)構(gòu)的同學(xué)應(yīng)該都知道，隊(duì)列是數(shù)據(jù)結(jié)構(gòu)中一種特殊的線(xiàn)性表結(jié)構(gòu)，和平時(shí)使用的List，Set這些數(shù)據(jù)結(jié)構(gòu)相比有點(diǎn)特殊，它的特殊之處在于它只允許在隊(duì)列的頭部（Head）進(jìn)行刪除操作，在尾部（Tail）進(jìn)行插入操作，這種方式的隊(duì)列我們稱(chēng)之為先進(jìn)先出隊(duì)列（FIFO）。

Java并發(fā)編程之阻塞隊(duì)列(BlockingQueue)詳解

在JDK1.5中推出了隊(duì)列這一數(shù)據(jù)結(jié)構(gòu)的具體實(shí)現(xiàn)，接口Queue是對(duì)于隊(duì)列的定義，并有一些列具有特殊功能的隊(duì)列實(shí)現(xiàn)。

在Queue接口中定義了隊(duì)列的如下方法：

Java并發(fā)編程之阻塞隊(duì)列(BlockingQueue)詳解

其中add(E)并非Queue接口新定義，而是從Collection接口繼承而來(lái)的。

阻塞隊(duì)列

BlockingQueue接口也是在JDK1.5中推出，存放在java.util.concurrent包中，繼承自Queue，所以在BlockingQueue中有Queue的所有方法。

從名字就可以看出BlockingQueue是一種阻塞隊(duì)列，它支持在檢索元素時(shí)如果隊(duì)列為空可以一直阻塞等待直到有元素可以獲取，同樣在添加元素時(shí)如果隊(duì)列已滿(mǎn)會(huì)阻塞等待隊(duì)列中有空閑的存儲(chǔ)空間。

BlockingQueue的方法可以歸納為四類(lèi)：

在操作時(shí)如不能立即滿(mǎn)足，會(huì)直接拋出異常
在操作時(shí)如不能立即滿(mǎn)足，則返回特殊的值，如插入、移除方法會(huì)返回false，檢查方法會(huì)返回null
在操作時(shí)如不能立即滿(mǎn)足，則會(huì)阻塞等待，直到操作成功
在操作時(shí)如不能立即滿(mǎn)足，則會(huì)阻塞等待給定的時(shí)間長(zhǎng)度，時(shí)間到達(dá)后如果還不能滿(mǎn)足則返回null

這四類(lèi)方法總結(jié)如下。

Java并發(fā)編程之阻塞隊(duì)列(BlockingQueue)詳解

因?yàn)樵贐lockingQueue的一些方法中，會(huì)通過(guò)null表示某種操作的失敗，所以不允許在BlockingQueue中存放null值元素，會(huì)在操作時(shí)拋出NullPointerExection異常。

BlockingQueue因?yàn)槭且粋€(gè)容器嘛，所以它也有容量的限制，在具體實(shí)現(xiàn)類(lèi)中有可以設(shè)置容量的實(shí)現(xiàn)類(lèi)，也有不可以設(shè)置容量的實(shí)現(xiàn)類(lèi)，不能設(shè)置容量的實(shí)現(xiàn)類(lèi)容量默認(rèn)為Integer.MAX_VALUE。

BlockingQueue是定義在java.util.concurrent包中，那么它在并發(fā)情況下到底是不是線(xiàn)程安全的呢？

在JDK提供的BlockingQueue的具體實(shí)現(xiàn)類(lèi)中，上面表格中的方法實(shí)現(xiàn)都是線(xiàn)程安全的，在內(nèi)部都使用了鎖或者其他形式的并發(fā)控制保證操作的原子性。

但是有一點(diǎn)要注意，就是一些批量處理的方法例如addAll、containsAll、retainAll和removeAll這些方法并不一定是線(xiàn)程安全的，使用時(shí)注意。

說(shuō)完BlockingQueue接口我們接下來(lái)看看它都有哪些具體的實(shí)現(xiàn)呢？以及在它們內(nèi)部是如何做到線(xiàn)程安全和阻塞的呢？

ArrayBlockingQueue

ArrayBlockingQueue是一個(gè)底層由數(shù)組支持額有界阻塞隊(duì)列。

重要屬性

先來(lái)看看ArrayBlockingQueue中都有哪些屬性。

// 存放元素的數(shù)組
final Object[] items;
// 用來(lái)記錄取元素的下標(biāo)，用于下一次在take，poll,remove，peek方法中使用
int takeIndex;
// 用來(lái)記錄添加元素的下標(biāo)，用于下一次put,offer,add等方法使用
int putIndex;
// 記錄隊(duì)列中元素?cái)?shù)量
int count;
// 用于控制并發(fā)訪(fǎng)問(wèn)時(shí)保證線(xiàn)程安全的鎖
final ReentrantLock lock;
// 用于隊(duì)列空時(shí)阻塞和喚醒等待線(xiàn)程的條件
private final Condition notEmpty;
// 用于隊(duì)列滿(mǎn)時(shí)阻塞和喚醒等待線(xiàn)程的條件
private final Condition notFull;

我們通過(guò)這些隊(duì)列中的屬性基本可以知道ArrayBlockingQueue中都有哪些重要信息，可以看出ArrayBlockingQueue就是使用Object[]來(lái)存放元素的。

那么應(yīng)該如何創(chuàng)建一個(gè)ArrayBlockingQueue呢？

構(gòu)造方法

public ArrayBlockingQueue(int capacity) {
  this(capacity, false);
}

默認(rèn)的構(gòu)造方法需要傳入一個(gè)int類(lèi)型的capacity表示該隊(duì)列的容量。在該構(gòu)造方法中會(huì)調(diào)用另一個(gè)構(gòu)造方法，傳入一個(gè)默認(rèn)值false。

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();
}

從這個(gè)方法我們看出傳入的false表示會(huì)在內(nèi)部用于創(chuàng)建一個(gè)ReentrantLock對(duì)象，我們都知道ReentrantLock支持公平和非公平的實(shí)現(xiàn)，我們猜想一下，這里的這個(gè)fair值是不是表示該阻塞隊(duì)列對(duì)于阻塞排隊(duì)的線(xiàn)程支持公平和非公平的策略呢？這里先賣(mài)個(gè)關(guān)子，在后面的方法中我們具體說(shuō)。

除了這兩種創(chuàng)建的方式，ArrayBlockingQueue還支持傳入一個(gè)Collection集合。

public ArrayBlockingQueue(int capacity, boolean fair,
                            Collection<? extends E> c) {
  // 先創(chuàng)建一個(gè)ArrayBlockingQueue實(shí)例
  this(capacity, fair);
  final ReentrantLock lock = this.lock;
  lock.lock();
  try {
      int i = 0;
      try {
          // 循環(huán)將collection中的元素放入queue中
          for (E e : c) {
              checkNotNull(e);
              items[i++] = e;
          }
      } catch (ArrayIndexOutOfBoundsException ex) {
          // 如果collection的元素個(gè)數(shù)超出queue的容量大小，會(huì)拋出異常
          throw new IllegalArgumentException();
      }
      count = i;
      putIndex = (i == capacity) ? 0 : i;
  } finally {
      lock.unlock();
  }
}

添加元素

先來(lái)看看添加一個(gè)新元素到ArrayBlockingQueue是如何實(shí)現(xiàn)的，怎樣保證線(xiàn)程安全的。

add(e)

public boolean add(E e) {
  // 調(diào)用父類(lèi)中的add(e)方法
  return super.add(e);
}
public boolean add(E e) {
  // 這里會(huì)直接調(diào)用offer(e)方法，如果offer方法返回false,則直接拋出異常
  if (offer(e))
      return true;
  else
      throw new IllegalStateException("Queue full");
}

add方法的實(shí)現(xiàn)邏輯本質(zhì)上是對(duì)offer方法套了一層殼，如果offer方法返回false時(shí)，拋出異常。所以我們直接看offer方法的實(shí)現(xiàn)就好。

offer(e)

public boolean offer(E e) {
  // 這里先判斷空，如果e為空會(huì)拋出空指針異常
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
	// 加鎖，保證入隊(duì)操作的原子性
  lock.lock();
  try {
      // 隊(duì)列滿(mǎn)時(shí)直接返回false
      if (count == items.length)
          return false;
      else {
          // 元素入隊(duì)
          enqueue(e);
          return true;
      }
  } finally {
      lock.unlock();
  }
}

可以看到offer方法的邏輯還是比較簡(jiǎn)單的，先檢查入?yún)⒉荒転榭?，然后加鎖保證入隊(duì)操作的原子性，在獲取鎖成功后入隊(duì)，如果隊(duì)列已滿(mǎn)則直接返回false，所以offer方法并不會(huì)阻塞。

put(e)

public void put(E e) throws InterruptedException {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
	// 可被中斷方式獲取鎖
  lock.lockInterruptibly();
  try {
      while (count == items.length)
          // 隊(duì)列滿(mǎn)時(shí)會(huì)阻塞
          notFull.await();
      // 入隊(duì)
      enqueue(e);
  } finally {
      lock.unlock();
  }
}

put方法和offer方法唯一的區(qū)別，就是會(huì)在隊(duì)列滿(mǎn)的時(shí)候使用Condition條件對(duì)象notFull阻塞等待。

private void enqueue(E x) {
  final Object[] items = this.items;
  items[putIndex] = x;
  if (++putIndex == items.length)
      putIndex = 0;
  count++;
  // 入隊(duì)成功，喚醒等待的移除元素操作線(xiàn)程
  notEmpty.signal();
}

在enqueue方法中才會(huì)完成對(duì)隊(duì)列中的數(shù)組元素的賦值動(dòng)作，完成之后喚醒阻塞等待的移除元素操作線(xiàn)程。

offer(e,time,unit)

public boolean offer(E e, long timeout, TimeUnit unit)
      throws InterruptedException {
  checkNotNull(e);
  // 加鎖之前先獲取需要等待的時(shí)間值
  long nanos = unit.toNanos(timeout);
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
      while (count == items.length) {
          // 時(shí)間小于等于0時(shí)，返回false
          if (nanos <= 0)
              return false;
          // 阻塞等待指定時(shí)間
          nanos = notFull.awaitNanos(nanos);
      }
      enqueue(e);
      return true;
  } finally {
      lock.unlock();
  }
}

offer(e,time,unit)方法與offer(e)方法相比，主要時(shí)多了一個(gè)等待時(shí)間，會(huì)在時(shí)間到達(dá)時(shí)如果沒(méi)有空間添加元素返回false。

移除元素

ArrayBlockingQueue中移除元素的方法主要有remove()，poll()，take()，poll(time,unit)四個(gè)。這幾個(gè)方法的實(shí)現(xiàn)邏輯都比較簡(jiǎn)單，這里不在單獨(dú)貼代碼。我們來(lái)看一下阻塞方法take()的實(shí)現(xiàn)即可。

take()

public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
	// 加鎖
  lock.lockInterruptibly();
  try {
      while (count == 0)
          // 如果元素?cái)?shù)量==0，表示隊(duì)列中為空，則阻塞等待
          notEmpty.await();
      return dequeue();
  } finally {
      lock.unlock();
  }
}

dequeue()

private E dequeue() {
  final Object[] items = this.items;
  @SuppressWarnings("unchecked")
  E x = (E) items[takeIndex];
  items[takeIndex] = null;
  if (++takeIndex == items.length)
      takeIndex = 0;
  count--;
  if (itrs != null)
      itrs.elementDequeued();
  // 取出元素之后，喚醒其他等待線(xiàn)程。
  notFull.signal();
  return x;
}

LinkedBlockingQueue

LinkedBlockingQueue是一個(gè)基于鏈表結(jié)構(gòu)的阻塞隊(duì)列，可以在創(chuàng)建時(shí)指定邊界大小，也可以不指定，在不指定邊界時(shí)容量為Integer.MAX_VALUE。

Java并發(fā)編程之阻塞隊(duì)列(BlockingQueue)詳解

重要屬性

我們先來(lái)看看在LinkedBlockingQueue中都有哪些重要的屬性。

// 內(nèi)部類(lèi)Node節(jié)點(diǎn)，用來(lái)存放鏈表中的元素
static class Node<E> {
  // 節(jié)點(diǎn)元素
	E item;
	// 當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)，如果為空表示沒(méi)有下一個(gè)節(jié)點(diǎn)
	Node<E> next;
	Node(E x) { item = x; }
}
// 隊(duì)列的容量
private final int capacity;
// 隊(duì)列中元素的數(shù)量
private final AtomicInteger count = new AtomicInteger();
// 頭節(jié)點(diǎn)
transient Node<E> head;
// 最后一個(gè)節(jié)點(diǎn)
private transient Node<E> last;
// 獲取元素時(shí)控制線(xiàn)程安全的鎖
private final ReentrantLock takeLock = new ReentrantLock();
// 添加元素時(shí)控制線(xiàn)程安全的鎖
private final ReentrantLock putLock = new ReentrantLock();
// 控制消費(fèi)者的條件
private final Condition notEmpty = takeLock.newCondition();
// 控制生產(chǎn)者的條件
private final Condition notFull = putLock.newCondition();

在LinkedBlockingQueue中使用Node來(lái)存放元素，和指向下一個(gè)節(jié)點(diǎn)的鏈表指針。

構(gòu)造方法

在LinkedBlockingQueue的構(gòu)造方法中，會(huì)創(chuàng)建一個(gè)創(chuàng)建一個(gè)不存放元素的Node對(duì)象賦值給head和last。

public LinkedBlockingQueue() {
  this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
  if (capacity <= 0) throw new IllegalArgumentException();
  this.capacity = capacity;
  // 創(chuàng)建一個(gè)不存放元素的Node對(duì)象賦值給head和last
  last = head = new Node<E>(null);
}
public LinkedBlockingQueue(Collection<? extends E> c) {
      this(Integer.MAX_VALUE);
      final ReentrantLock putLock = this.putLock;
      putLock.lock();
      try {
          int n = 0;
          for (E e : c) {
              if (e == null)
                  throw new NullPointerException();
              if (n == capacity)
                  throw new IllegalStateException("Queue full");
              // 入隊(duì)
              enqueue(new Node<E>(e));
              ++n;
          }
          count.set(n);
      } finally {
          putLock.unlock();
      }
  }

添加元素

offer(e)

public boolean offer(E e) {
  if (e == null) throw new NullPointerException();
  final AtomicInteger count = this.count;
  if (count.get() == capacity)
      return false;
  int c = -1;
  Node<E> node = new Node<E>(e);
  // 使用putLock加鎖
  final ReentrantLock putLock = this.putLock;
  putLock.lock();
  try {
      if (count.get() < capacity) {
          // 入隊(duì)
          enqueue(node);
          // 數(shù)量+1
          c = count.getAndIncrement();
          if (c + 1 < capacity)
              // 喚醒一個(gè)生產(chǎn)者線(xiàn)程
              notFull.signal();
      }
  } finally {
      putLock.unlock();
  }
  if (c == 0)
      // 喚醒消費(fèi)者線(xiàn)程
      signalNotEmpty();
  // 入隊(duì)失敗情況會(huì)返回false
  return c >= 0;
}

對(duì)于鏈表結(jié)構(gòu)的LinkedBlockingQueue來(lái)說(shuō)，入隊(duì)操作要簡(jiǎn)單很多，只需要將node節(jié)點(diǎn)掛在最后一個(gè)節(jié)點(diǎn)last的next，然后將自己賦值給last。

private void enqueue(Node<E> node) {
  last = last.next = node;
}

put(e)

public void put(E e) throws InterruptedException {
  if (e == null) throw new NullPointerException();
  int c = -1;
  Node<E> node = new Node<E>(e);
  // 使用putLock加鎖
  final ReentrantLock putLock = this.putLock;
  final AtomicInteger count = this.count;
  putLock.lockInterruptibly();
  try {
      while (count.get() == capacity) {
          // 如果隊(duì)列容量已使用完則阻塞
          notFull.await();
      }
      enqueue(node);
      c = count.getAndIncrement();
      if (c + 1 < capacity)
          notFull.signal();
  } finally {
      putLock.unlock();
  }
  if (c == 0)
      signalNotEmpty();
}

對(duì)比結(jié)果也和我們最開(kāi)始的方法匯總表格一樣，offer(e)方法會(huì)在入隊(duì)時(shí)如果隊(duì)列已滿(mǎn)直接返回false，而put(e)會(huì)一直阻塞等待，知道入隊(duì)成功。

add(e)方法和offer(e,time,unit)方法實(shí)現(xiàn)邏輯上沒(méi)有特殊之處，這里不再放源碼。

移除元素

poll()

public E poll() {
      final AtomicInteger count = this.count;
      if (count.get() == 0)
          return null;
      E x = null;
      int c = -1;
  // 使用takeLock加鎖
      final ReentrantLock takeLock = this.takeLock;
      takeLock.lock();
      try {
          if (count.get() > 0) {
              x = dequeue();
              c = count.getAndDecrement();
              if (c > 1)
                  // 還有元素時(shí)喚醒一個(gè)生產(chǎn)者線(xiàn)程
                  notEmpty.signal();
          }
      } finally {
          takeLock.unlock();
      }
      if (c == capacity)
          // 喚醒生產(chǎn)者線(xiàn)程
          signalNotFull();
      return x;
  }

poll()方法會(huì)在元素出隊(duì)時(shí)如果沒(méi)有元素則直接返回null。

// 出隊(duì)方法
private E dequeue() {
  Node<E> h = head;
  Node<E> first = h.next;
  h.next = h; 
  head = first;
  E x = first.item;
  first.item = null;
  return x;
}

take()

public E take() throws InterruptedException {
  E x;
  int c = -1;
  final AtomicInteger count = this.count;
  // 使用takeLock加鎖
  final ReentrantLock takeLock = this.takeLock;
  takeLock.lockInterruptibly();
  try {
      while (count.get() == 0) {
          //阻塞等待
          notEmpty.await();
      }
      x = dequeue();
      c = count.getAndDecrement();
      if (c > 1)
          // 還有元素時(shí)喚醒一個(gè)消費(fèi)者線(xiàn)程
          notEmpty.signal();
  } finally {
      takeLock.unlock();
  }
  if (c == capacity)
      // 喚醒生產(chǎn)者線(xiàn)程
      signalNotFull();
  return x;
}

同樣，take方法會(huì)在沒(méi)有元素時(shí)一直等待。

對(duì)比

我們來(lái)對(duì)比一下ArrayBlockingQueue和LinkedBlockingQueue都有哪些區(qū)別。

ArrayBlockingQueue基于數(shù)組實(shí)現(xiàn)，LinkedBlockingQueue基于鏈表實(shí)現(xiàn)
ArrayBlockingQueue在添加和移除元素的操作中共用一把鎖，LinkedBlockingQueue使用takeLock和putLock兩把鎖
ArrayBlockingQueue在添加和移除元素時(shí)直接使用元素的類(lèi)型處理，LinkedBlockingQueue需要轉(zhuǎn)成Node對(duì)象
ArrayBlockingQueue創(chuàng)建時(shí)必須指定容量，LinkedBlockingQueue可以不指定，默認(rèn)容量為Integer.MAX_VALUE

由于LinkedBlockingQueue使用兩把鎖將入隊(duì)操作和出隊(duì)操作分離，這會(huì)大大提高隊(duì)列的吞吐量，在高并發(fā)情況下生產(chǎn)者和消費(fèi)者可以并行處理，提高并發(fā)性能。

但是LinkedBlockingQueue默認(rèn)是無(wú)界隊(duì)列，要小心內(nèi)存溢出風(fēng)險(xiǎn)，所以最好在創(chuàng)建時(shí)指定容量大小。

BlockingQueue接口的實(shí)現(xiàn)類(lèi)除了本期介紹的這兩種，還有PriorityBlockingQueue，SynchronousQueue，LinkedBlockingDeque等，每一個(gè)都有它獨(dú)特的特性和使用場(chǎng)景，后面我們?cè)賳为?dú)深入解析。