国产探花免费观看_亚洲丰满少妇自慰呻吟_97日韩有码在线_资源在线日韩欧美_一区二区精品毛片,辰东完美世界有声小说,欢乐颂第一季,yy玄幻小说排行榜完本

首頁 > 編程 > Java > 正文

剖析Java中阻塞隊列的實現原理及應用場景

2019-11-26 14:46:46
字體:
來源:轉載
供稿:網友

我們平時使用的一些常見隊列都是非阻塞隊列,比如PriorityQueue、LinkedList(LinkedList是雙向鏈表,它實現了Dequeue接口)。
使用非阻塞隊列的時候有一個很大問題就是:它不會對當前線程產生阻塞,那么在面對類似消費者-生產者的模型時,就必須額外地實現同步策略以及線程間喚醒策略,這個實現起來就非常麻煩。但是有了阻塞隊列就不一樣了,它會對當前線程產生阻塞,比如一個線程從一個空的阻塞隊列中取元素,此時線程會被阻塞直到阻塞隊列中有了元素。當隊列中有元素后,被阻塞的線程會自動被喚醒(不需要我們編寫代碼去喚醒)。這樣提供了極大的方便性。
一.幾種主要的阻塞隊列

  自從Java 1.5之后,在java.util.concurrent包下提供了若干個阻塞隊列,主要有以下幾個:

  ArrayBlockingQueue:基于數組實現的一個阻塞隊列,在創建ArrayBlockingQueue對象時必須制定容量大小。并且可以指定公平性與非公平性,默認情況下為非公平的,即不保證等待時間最長的隊列最優先能夠訪問隊列。

  LinkedBlockingQueue:基于鏈表實現的一個阻塞隊列,在創建LinkedBlockingQueue對象時如果不指定容量大小,則默認大小為Integer.MAX_VALUE。

  PriorityBlockingQueue:以上2種隊列都是先進先出隊列,而PriorityBlockingQueue卻不是,它會按照元素的優先級對元素進行排序,按照優先級順序出隊,每次出隊的元素都是優先級最高的元素。注意,此阻塞隊列為無界阻塞隊列,即容量沒有上限(通過源碼就可以知道,它沒有容器滿的信號標志),前面2種都是有界隊列。

  DelayQueue:基于PriorityQueue,一種延時阻塞隊列,DelayQueue中的元素只有當其指定的延遲時間到了,才能夠從隊列中獲取到該元素。DelayQueue也是一個無界隊列,因此往隊列中插入數據的操作(生產者)永遠不會被阻塞,而只有獲取數據的操作(消費者)才會被阻塞。

二.阻塞隊列中的方法 VS 非阻塞隊列中的方法

1.非阻塞隊列中的幾個主要方法:

  •   add(E e):將元素e插入到隊列末尾,如果插入成功,則返回true;如果插入失敗(即隊列已滿),則會拋出異常;
  •   remove():移除隊首元素,若移除成功,則返回true;如果移除失敗(隊列為空),則會拋出異常;
  •   offer(E e):將元素e插入到隊列末尾,如果插入成功,則返回true;如果插入失敗(即隊列已滿),則返回false;
  •   poll():移除并獲取隊首元素,若成功,則返回隊首元素;否則返回null;
  •   peek():獲取隊首元素,若成功,則返回隊首元素;否則返回null

 

  對于非阻塞隊列,一般情況下建議使用offer、poll和peek三個方法,不建議使用add和remove方法。因為使用offer、poll和peek三個方法可以通過返回值判斷操作成功與否,而使用add和remove方法卻不能達到這樣的效果。注意,非阻塞隊列中的方法都沒有進行同步措施。

2.阻塞隊列中的幾個主要方法:

  阻塞隊列包括了非阻塞隊列中的大部分方法,上面列舉的5個方法在阻塞隊列中都存在,但是要注意這5個方法在阻塞隊列中都進行了同步措施。除此之外,阻塞隊列提供了另外4個非常有用的方法:

  1.   put(E e)
  2.   take()
  3.   offer(E e,long timeout, TimeUnit unit)
  4.   poll(long timeout, TimeUnit unit)

  

  •   put方法用來向隊尾存入元素,如果隊列滿,則等待;
  •   take方法用來從隊首取元素,如果隊列為空,則等待;
  •   offer方法用來向隊尾存入元素,如果隊列滿,則等待一定的時間,當時間期限達到時,如果還沒有插入成功,則返回false;否則返回true;
  •   poll方法用來從隊首取元素,如果隊列空,則等待一定的時間,當時間期限達到時,如果取到,則返回null;否則返回取得的元素;

三.阻塞隊列的實現原理

如果隊列是空的,消費者會一直等待,當生產者添加元素時候,消費者是如何知道當前隊列有元素的呢?如果讓你來設計阻塞隊列你會如何設計,讓生產者和消費者能夠高效率的進行通訊呢?讓我們先來看看JDK是如何實現的。

使用通知模式實現。所謂通知模式,就是當生產者往滿的隊列里添加元素時會阻塞住生產者,當消費者消費了一個隊列中的元素后,會通知生產者當前隊列可用。通過查看JDK源碼發現ArrayBlockingQueue使用了Condition來實現,代碼如下:

private final Condition notFull;private final Condition notEmpty;public ArrayBlockingQueue(int capacity, boolean fair) {    //省略其他代碼    notEmpty = lock.newCondition();    notFull = lock.newCondition();  }public void put(E e) throws InterruptedException {    checkNotNull(e);    final ReentrantLock lock = this.lock;    lock.lockInterruptibly();    try {      while (count == items.length)        notFull.await();      insert(e);    } finally {      lock.unlock();    }}public E take() throws InterruptedException {    final ReentrantLock lock = this.lock;    lock.lockInterruptibly();    try {      while (count == 0)        notEmpty.await();      return extract(); } finally {      lock.unlock();    }}private void insert(E x) {    items[putIndex] = x;    putIndex = inc(putIndex);    ++count;    notEmpty.signal();  }

當我們往隊列里插入一個元素時,如果隊列不可用,阻塞生產者主要通過LockSupport.park(this);來實現

public final void await() throws InterruptedException {      if (Thread.interrupted())        throw new InterruptedException();      Node node = addConditionWaiter();      int savedState = fullyRelease(node);      int interruptMode = 0;      while (!isOnSyncQueue(node)) {        LockSupport.park(this);        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)          break;      }      if (acquireQueued(node, savedState) && interruptMode != THROW_IE)        interruptMode = REINTERRUPT;      if (node.nextWaiter != null) // clean up if cancelled        unlinkCancelledWaiters();      if (interruptMode != 0)reportInterruptAfterWait(interruptMode);    }

繼續進入源碼,發現調用setBlocker先保存下將要阻塞的線程,然后調用unsafe.park阻塞當前線程。

public static void park(Object blocker) {    Thread t = Thread.currentThread();    setBlocker(t, blocker);    unsafe.park(false, 0L);    setBlocker(t, null);  }

unsafe.park是個native方法,代碼如下:

public native void park(boolean isAbsolute, long time);

park這個方法會阻塞當前線程,只有以下四種情況中的一種發生時,該方法才會返回。

與park對應的unpark執行或已經執行時。注意:已經執行是指unpark先執行,然后再執行的park。
線程被中斷時。
如果參數中的time不是零,等待了指定的毫秒數時。
發生異常現象時。這些異常事先無法確定。
我們繼續看一下JVM是如何實現park方法的,park在不同的操作系統使用不同的方式實現,在linux下是使用的是系統方法pthread_cond_wait實現。實現代碼在JVM源碼路徑src/os/linux/vm/os_linux.cpp里的 os::PlatformEvent::park方法,代碼如下:

void os::PlatformEvent::park() {         int v ;   for (;;) { v = _Event ;   if (Atomic::cmpxchg (v-1, &_Event, v) == v) break ;   }   guarantee (v >= 0, "invariant") ;   if (v == 0) {   // Do this the hard way by blocking ...   int status = pthread_mutex_lock(_mutex);   assert_status(status == 0, status, "mutex_lock");   guarantee (_nParked == 0, "invariant") ;   ++ _nParked ;   while (_Event < 0) {   status = pthread_cond_wait(_cond, _mutex);   // for some reason, under 2.7 lwp_cond_wait() may return ETIME ...   // Treat this the same as if the wait was interrupted   if (status == ETIME) { status = EINTR; }   assert_status(status == 0 || status == EINTR, status, "cond_wait");   }   -- _nParked ;      // In theory we could move the ST of 0 into _Event past the unlock(),   // but then we'd need a MEMBAR after the ST.   _Event = 0 ;   status = pthread_mutex_unlock(_mutex);   assert_status(status == 0, status, "mutex_unlock");   }   guarantee (_Event >= 0, "invariant") ;   }   }

pthread_cond_wait是一個多線程的條件變量函數,cond是condition的縮寫,字面意思可以理解為線程在等待一個條件發生,這個條件是一個全局變量。這個方法接收兩個參數,一個共享變量_cond,一個互斥量_mutex。而unpark方法在linux下是使用pthread_cond_signal實現的。park 在windows下則是使用WaitForSingleObject實現的。

當隊列滿時,生產者往阻塞隊列里插入一個元素,生產者線程會進入WAITING (parking)狀態。我們可以使用jstack dump阻塞的生產者線程看到這點:

"main" prio=5 tid=0x00007fc83c000000 nid=0x10164e000 waiting on condition [0x000000010164d000]  java.lang.Thread.State: WAITING (parking)    at sun.misc.Unsafe.park(Native Method)    - parking to wait for <0x0000000140559fe8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)    at java.util.concurrent.ArrayBlockingQueue.put(ArrayBlockingQueue.java:324)    at blockingqueue.ArrayBlockingQueueTest.main(ArrayBlockingQueueTest.java:11)

四.示例和使用場景

  下面先使用Object.wait()和Object.notify()、非阻塞隊列實現生產者-消費者模式:

public class Test {  private int queueSize = 10;  private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);     public static void main(String[] args) {    Test test = new Test();    Producer producer = test.new Producer();    Consumer consumer = test.new Consumer();         producer.start();    consumer.start();  }     class Consumer extends Thread{         @Override    public void run() {      consume();    }         private void consume() {      while(true){        synchronized (queue) {          while(queue.size() == 0){            try {              System.out.println("隊列空,等待數據");              queue.wait();            } catch (InterruptedException e) {              e.printStackTrace();              queue.notify();            }          }          queue.poll();     //每次移走隊首元素          queue.notify();          System.out.println("從隊列取走一個元素,隊列剩余"+queue.size()+"個元素");        }      }    }  }     class Producer extends Thread{         @Override    public void run() {      produce();    }         private void produce() {      while(true){        synchronized (queue) {          while(queue.size() == queueSize){            try {              System.out.println("隊列滿,等待有空余空間");              queue.wait();            } catch (InterruptedException e) {              e.printStackTrace();              queue.notify();            }          }          queue.offer(1);    //每次插入一個元素          queue.notify();          System.out.println("向隊列取中插入一個元素,隊列剩余空間:"+(queueSize-queue.size()));        }      }    }  }}

   這個是經典的生產者-消費者模式,通過阻塞隊列和Object.wait()和Object.notify()實現,wait()和notify()主要用來實現線程間通信。

  具體的線程間通信方式(wait和notify的使用)在后續問章中會講述到。

  下面是使用阻塞隊列實現的生產者-消費者模式:

public class Test {  private int queueSize = 10;  private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(queueSize);     public static void main(String[] args) {    Test test = new Test();    Producer producer = test.new Producer();    Consumer consumer = test.new Consumer();         producer.start();    consumer.start();  }     class Consumer extends Thread{         @Override    public void run() {      consume();    }         private void consume() {      while(true){        try {          queue.take();          System.out.println("從隊列取走一個元素,隊列剩余"+queue.size()+"個元素");        } catch (InterruptedException e) {          e.printStackTrace();        }      }    }  }     class Producer extends Thread{         @Override    public void run() {      produce();    }         private void produce() {      while(true){        try {          queue.put(1);          System.out.println("向隊列取中插入一個元素,隊列剩余空間:"+(queueSize-queue.size()));        } catch (InterruptedException e) {          e.printStackTrace();        }      }    }  }}

   有沒有發現,使用阻塞隊列代碼要簡單得多,不需要再單獨考慮同步和線程間通信的問題。

  在并發編程中,一般推薦使用阻塞隊列,這樣實現可以盡量地避免程序出現意外的錯誤。

  阻塞隊列使用最經典的場景就是socket客戶端數據的讀取和解析,讀取數據的線程不斷將數據放入隊列,然后解析線程不斷從隊列取數據解析。還有其他類似的場景,只要符合生產者-消費者模型的都可以使用阻塞隊列。

發表評論 共有條評論
用戶名: 密碼:
驗證碼: 匿名發表
主站蜘蛛池模板: 桑植县| 洪雅县| 巩留县| 富蕴县| 枣阳市| 普洱| 清水河县| 鄂尔多斯市| 南郑县| 芒康县| 寻乌县| 河津市| 朔州市| 舒兰市| 东阳市| 巫山县| 梁山县| 鸡东县| 隆德县| 崇明县| 子洲县| 上思县| 奉化市| 库尔勒市| 佛冈县| 罗源县| 宽城| 忻州市| 肇源县| 伊金霍洛旗| 民丰县| 海伦市| 贵德县| 武山县| 开阳县| 武宁县| 彩票| 榆树市| 新巴尔虎右旗| 梨树县| 五台县|