解析java中的condition

一、condition 介紹及demo

Condition是在java 1.5中才出現的,它用來替代傳統的Object的wait()、notify()實現線程間的協作,相比使用Object的wait()、notify(),使用Condition的await()、signal()這種方式實現線程間協作更加安全和高效。因此通常來說比較推薦使用Condition,阻塞隊列實際上是使用瞭Condition來模擬線程間協作。

  • Condition是個接口,基本的方法就是await()和signal()方法;
  • Condition依賴於Lock接口,生成一個Condition的基本代碼是lock.newCondition()   
  • 調用Condition的await()和signal()方法,都必須在lock保護之內,就是說必須在lock.lock()和lock.unlock之間才可以使用

Conditon中的await()對應Object的wait();

Condition中的signal()對應Object的notify();

Condition中的signalAll()對應Object的notifyAll()。

condition常見例子arrayblockingqueue。下面是demo:

package thread;
 
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ConTest {
	
	 final Lock lock = new ReentrantLock();
	 final Condition condition = lock.newCondition();
 
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		ConTest test = new ConTest();
	    Producer producer = test.new Producer();
	    Consumer consumer = test.new Consumer();
	          
	    
	    consumer.start(); 
	    producer.start();
	}
	
	 class Consumer extends Thread{
         
	        @Override
	        public void run() {
	            consume();
	        }
	          
	        private void consume() {
	           	             
	                try {
	                	   lock.lock();
	                    System.out.println("我在等一個新信號"+this.currentThread().getName());
	                    condition.await();
	                    
	                } catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} finally{
						System.out.println("拿到一個信號"+this.currentThread().getName());
	                    lock.unlock();
	                }
	            
	        }
	    }
	 
	 class Producer extends Thread{
         
	        @Override
	        public void run() {
	            produce();
	        }
	          
	        private void produce() {	             
	                try {
	                	   lock.lock();
	                       System.out.println("我拿到鎖"+this.currentThread().getName());
	                        condition.signalAll();	                         
	                    System.out.println("我發出瞭一個信號:"+this.currentThread().getName());
	                } finally{
	                    lock.unlock();
	                }
	            }
	 }
	    
}

運行結果:

Condition的執行方式,是當在線程Consumer中調用await方法後,線程Consumer將釋放鎖,並且將自己沉睡,等待喚醒,線程Producer獲取到鎖後,開始做事,完畢後,調用Condition的signalall方法,喚醒線程Consumer,線程Consumer恢復執行。

以上說明Condition是一個多線程間協調通信的工具類,使得某個,或者某些線程一起等待某個條件(Condition),隻有當該條件具備( signal 或者 signalAll方法被帶調用)時 ,這些等待線程才會被喚醒,從而重新爭奪鎖。

Condition實現生產者、消費者模式:

package thread;

import java.util.PriorityQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConTest2 {
    private int queueSize = 10;
    private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
    private Lock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();
    private Condition notEmpty = lock.newCondition();
        
    public static void main(String[] args) throws InterruptedException  {
        ConTest2 test = new ConTest2();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();	          
        producer.start();
        consumer.start();
        Thread.sleep(0);
        producer.interrupt();
        consumer.interrupt();
    }
        
    class Consumer extends Thread{	          
        @Override
        public void run() {
            consume();
        }
        volatile boolean flag=true;  
        private void consume() {
            while(flag){
                lock.lock();
                try {
                    while(queue.isEmpty()){
                        try {
                            System.out.println("隊列空,等待數據");
                            notEmpty.await();
                        } catch (InterruptedException e) {	                          
                            flag =false;
                        }
                    }
                    queue.poll();                //每次移走隊首元素
                    notFull.signal();
                    System.out.println("從隊列取走一個元素,隊列剩餘"+queue.size()+"個元素");
                } finally{
                    lock.unlock();
                }
            }
        }
    }
        
    class Producer extends Thread{	          
        @Override
        public void run() {
            produce();
        }
        volatile boolean flag=true;  
        private void produce() {
            while(flag){
                lock.lock();
                try {
                    while(queue.size() == queueSize){
                        try {
                            System.out.println("隊列滿,等待有空餘空間");
                            notFull.await();
                        } catch (InterruptedException e) {
                            
                            flag =false;
                        }
                    }
                    queue.offer(1);        //每次插入一個元素
                    notEmpty.signal();
                    System.out.println("向隊列取中插入一個元素,隊列剩餘空間:"+(queueSize-queue.size()));
                } finally{
                    lock.unlock();
                }
            }
        }
    }
}

運行結果如下:

二、Condition接口     

condition可以通俗的理解為條件隊列。當一個線程在調用瞭await方法以後,直到線程等待的某個條件為真的時候才會被喚醒。這種方式為線程提供瞭更加簡單的等待/通知模式。Condition必須要配合鎖一起使用,因為對共享狀態變量的訪問發生在多線程環境下。一個Condition的實例必須與一個Lock綁定,因此Condition一般都是作為Lock的內部實現。

await() :造成當前線程在接到信號或被中斷之前一直處於等待狀態。

await(long time, TimeUnit unit) :造成當前線程在接到信號、被中斷或到達指定等待時間之前一直處於等待狀態

awaitNanos(long nanosTimeout) :造成當前線程在接到信號、被中斷或到達指定等待時間之前一直處於等待狀態。

返回值表示剩餘時間,如果在nanosTimesout之前喚醒,那麼返回值 = nanosTimeout – 消耗時間,如果返回值 <= 0 ,則可以認定它已經超時瞭。

awaitUninterruptibly() :造成當前線程在接到信號之前一直處於等待狀態。【註意:該方法對中斷不敏感】。

awaitUntil(Date deadline) :造成當前線程在接到信號、被中斷或到達指定最後期限之前一直處於等待狀態。如果沒有到指定時間就被通知,則返回true,否則表示到瞭指定時間,返回返回false。

signal() :喚醒一個等待線程。該線程從等待方法返回前必須獲得與Condition相關的鎖。

signal()All :喚醒所有等待線程。能夠從等待方法返回的線程必須獲得與Condition相關的鎖。

三、condition實現分析

  • Condition接口包含瞭多種await方式和兩個通知方法
  • ConditionObject實現瞭Condition接口,是AbstractQueuedSynchronizer的內部類(因為Condition的操作都需要獲取想關聯的鎖)
  • Reentrantlock的newCondition方法返回與某個lock實例相關的Condition對象
public abstract class AbstractQueuedLongSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

結合上面的類圖,我們看到condition實現是依賴於aqs,而aqs是個抽象類。裡面定義瞭同步器的基本框架,實現瞭基本的結構功能。隻留有狀態條件的維護由具體同步器根據具體場景來定制,如常見的 ReentrantLock 、 RetrantReadWriteLock和CountDownLatch 等等,

3.1、等待隊列

Condition是AQS的內部類。每個Condition對象都包含一個隊列(等待隊列)。等待隊列是一個FIFO的隊列,在隊列中的每個節點都包含瞭一個線程引用,該線程就是在Condition對象上等待的線程,如果一個線程調用瞭Condition.await()方法,那麼該線程將會釋放鎖、構造成節點加入等待隊列並進入等待狀態。AQS有一個同步隊列和多個等待隊列,節點都是Node。等待隊列的基本結構如下所示。

等待分為首節點和尾節點。當一個線程調用Condition.await()方法,將會以當前線程構造節點,並將節點從尾部加入等待隊列。新增節點就是將尾部節點指向新增的節點。節點引用更新本來就是在獲取鎖以後的操作,所以不需要CAS保證。同時也是線程安全的操作。

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    /** First node of condition queue. */
    private transient Node firstWaiter;
    /** Last node of condition queue. */
    private transient Node lastWaiter;

3.2、等待

 當線程調用瞭Condition的await()方法以後。線程就作為隊列中的一個節點被加入到等待隊列中去瞭。同時會釋放鎖的擁有。當從await方法返回的時候。當前線程一定會獲取condition相關聯的鎖。

如果從隊列(同步隊列和等待隊列)的角度去看await()方法,當調用await()方法時,相當於同步隊列的首節點(獲取鎖的節點)移動到Condition的等待隊列中。

調用該方法的線程成功的獲取鎖的線程,也就是同步隊列的首節點,該方法會將當前線程構造成節點並加入到等待隊列中,然後釋放同步狀態,喚醒同步隊列中的後繼節點,然後當前線程會進入等待狀態。

當等待隊列中的節點被喚醒的時候,則喚醒節點的線程開始嘗試獲取同步狀態。如果不是通過 其他線程調用Condition.signal()方法喚醒,而是對等待線程進行中斷,則會拋出InterruptedException異常信息。

   

我們看一下這個await的方法,它是AQS的方法,

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter(); //將當前線程包裝下後,
    //添加到Condition自己維護的一個鏈表中。
    int savedState = fullyRelease(node);//釋放當前線程占有的鎖,從demo中看到,
    //調用await前,當前線程是占有鎖的
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        //釋放完畢後,遍歷AQS的隊列,看當前節點是否在隊列中,
        //不在 說明它還沒有競爭鎖的資格,所以繼續將自己沉睡。
        //直到它被加入到隊列中,聰明的你可能猜到瞭,
        //沒有錯,在singal的時候加入不就可以瞭?
    LockSupport.park(this);
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
    }  
//被喚醒後,重新開始正式競爭鎖,同樣,如果競爭不到還是會將自己沉睡,等待喚醒重新開始競爭。
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
    unlinkCancelledWaiters();
if (interruptMode != 0)
    reportInterruptAfterWait(interruptMode);
}

結合代碼去看,同步隊列的首節點 並不會直接加入等待隊列,而是通過addConditionWaiter把當前線程構造成一個新節點並加入到等待隊列中。

/**
    * Adds a new waiter to wait queue.
    * @return its new wait node
    */
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

3.3、通知

 調用Condition的signal()方法,將會喚醒在等待隊列中等待最長時間的節點(條件隊列裡的首節點),在喚醒節點前,會將節點移到同步隊列中。當前線程加入到等待隊列中如圖所示:

回到上面的demo,鎖被釋放後,線程Consumer開始沉睡,這個時候線程因為線程Consumer沉睡時,會喚醒AQS隊列中的頭結點,所所以線程Producer會開始競爭鎖,並獲取到,執行完後線程Producer會調用signal方法,“發出”signal信號,signal方法如下:

public final void signal() {
 if (!isHeldExclusively())
    throw new IllegalMonitorStateException();
 Node first = firstWaiter; //firstWaiter為condition自己維護的一個鏈表的頭結點,
 //取出第一個節點後開始喚醒操作
 if (first != null)
    doSignal(first);
}

在調用signal()方法之前必須先判斷是否獲取到瞭鎖(isHeldExclusively方法)。接著獲取等待隊列的首節點,將其移動到同步隊列並且利用LockSupport喚醒節點中的線程。

被喚醒的線程將從await方法中的while循環中退出(  while (!isOnSyncQueue(node)) { 方法返回true,節點已經在同步隊列中)。隨後調用同步器的acquireQueued()方法加入到同步狀態的競爭當中去。成功獲取到競爭的線程從先前調用await方法返回,此時該線程已經成功獲取瞭鎖。

*********************************************

AQS的同步隊列與Condition的等待隊列,兩個隊列的作用是不同,事實上,每個線程也僅僅會同時存在以上兩個隊列中的一個,流程是這樣的:

註意:

1.線程producer調用signal方法,這個時候Condition的等待隊列中隻有線程Consumer一個節點,於是它被取出來,並被加入到AQS的等待隊列中。  註意,這個時候,線程Consumer 並沒有被喚醒。

2.Sync是AQS的抽象子類,實現可重入和互斥的大部分功能。在Sync的子類中有FairSync和NonfairSync兩種代表公平鎖策略和非公平鎖策略。Sync lock方法留給子類去實現,NonfairSync的實現:

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

其中如果一開始獲取鎖成功,是直接設置當前線程。

否則執行acquire(1),也就是進入aqs等待隊列。這裡不展開細節。

可以這樣理解,整個協作過程是靠結點在AQS的等待隊列和Condition的等待隊列中來回移動實現的,每個隊列的意義不同,Condition作為一個條件類,很好的自己維護瞭一個等待信號的隊列,並在適時的時候將結點加入到AQS的等待隊列中來實現的喚醒操作

以上就是解析java中的condition的詳細內容,更多關於java condition的資料請關註WalkonNet其它相關文章!

推薦閱讀: