我們平時使用的一些常見隊列都是非阻塞隊列,比如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.非阻塞隊列中的幾個主要方法:
對于非阻塞隊列,一般情況下建議使用offer、poll和peek三個方法,不建議使用add和remove方法。因為使用offer、poll和peek三個方法可以通過返回值判斷操作成功與否,而使用add和remove方法卻不能達到這樣的效果。注意,非阻塞隊列中的方法都沒有進行同步措施。
2.阻塞隊列中的幾個主要方法:
阻塞隊列包括了非阻塞隊列中的大部分方法,上面列舉的5個方法在阻塞隊列中都存在,但是要注意這5個方法在阻塞隊列中都進行了同步措施。除此之外,阻塞隊列提供了另外4個非常有用的方法:
三.阻塞隊列的實現原理
如果隊列是空的,消費者會一直等待,當生產者添加元素時候,消費者是如何知道當前隊列有元素的呢?如果讓你來設計阻塞隊列你會如何設計,讓生產者和消費者能夠高效率的進行通訊呢?讓我們先來看看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客戶端數據的讀取和解析,讀取數據的線程不斷將數據放入隊列,然后解析線程不斷從隊列取數據解析。還有其他類似的場景,只要符合生產者-消費者模型的都可以使用阻塞隊列。
新聞熱點
疑難解答