package thread;
 
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 
 * @author zhangliang
 *
 * 2016年4月8日 下午5:48:54
 */
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("我在等一個(gè)新信號(hào)"+this.currentThread().getName());
	                    condition.await();
	                    
	                } catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} finally{
						System.out.println("拿到一個(gè)信號(hào)"+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("我發(fā)出了一個(gè)信號(hào)："+this.currentThread().getName());
	                } finally{
	                    lock.unlock();
	                }
	            }
	 }
	    
}

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("隊(duì)列空，等待數(shù)據(jù)");
	                            notEmpty.await();
	                        } catch (InterruptedException e) {	                          
	                            flag =false;
	                        }
	                    }
	                    queue.poll();                //每次移走隊(duì)首元素
	                    notFull.signal();
	                    System.out.println("從隊(duì)列取走一個(gè)元素，隊(duì)列剩余"+queue.size()+"個(gè)元素");
	                } 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("隊(duì)列滿(mǎn)，等待有空余空間");
	                            notFull.await();
	                        } catch (InterruptedException e) {
	                            
	                            flag =false;
	                        }
	                    }
	                    queue.offer(1);        //每次插入一個(gè)元素
	                    notEmpty.signal();
	                    System.out.println("向隊(duì)列取中插入一個(gè)元素，隊(duì)列剩余空間："+(queueSize-queue.size()));
	                } finally{
	                    lock.unlock();
	                }
	            }
	        }
	    }
	}

public abstract class AbstractQueuedLongSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

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;

public final void await() throws InterruptedException {
	if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter(); //將當(dāng)前線程包裝下后，
//添加到Condition自己維護(hù)的一個(gè)鏈表中。
int savedState = fullyRelease(node);//釋放當(dāng)前線程占有的鎖，從demo中看到，
//調(diào)用await前，當(dāng)前線程是占有鎖的

int interruptMode = 0;
	 while (!isOnSyncQueue(node)) {//釋放完畢后，遍歷AQS的隊(duì)列，看當(dāng)前節(jié)點(diǎn)是否在隊(duì)列中，
//不在 說(shuō)明它還沒(méi)有競(jìng)爭(zhēng)鎖的資格，所以繼續(xù)將自己沉睡。
//直到它被加入到隊(duì)列中，聰明的你可能猜到了，
//沒(méi)有錯(cuò)，在singal的時(shí)候加入不就可以了？
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//被喚醒后，重新開(kāi)始正式競(jìng)爭(zhēng)鎖，同樣，如果競(jìng)爭(zhēng)不到還是會(huì)將自己沉睡，等待喚醒重新開(kāi)始競(jìng)爭(zhēng)。
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}

/**
         * 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;
        }

public final void signal() {
	 if (!isHeldExclusively())
	 throw new IllegalMonitorStateException();
	 Node first = firstWaiter; //firstWaiter為condition自己維護(hù)的一個(gè)鏈表的頭結(jié)點(diǎn)，
	                          //取出第一個(gè)節(jié)點(diǎn)后開(kāi)始喚醒操作
	 if (first != null)
	 doSignal(first);
	 }

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