Java多线程Condition接口原理介绍

Lock lock = new ReentrantLock();
 Condition condition = lock.newCondition();
 public void conditionWait() throws InterruptedException {
  lock.lock();
  try {
   condition.await();
  } finally {
   lock.unlock();
  }
 }
 public void conditionSignal() throws InterruptedException {
  lock.lock();
  try {
   condition.signal();
  } finally {
   lock.unlock();
  }
 }

 /**
  * 当前线程进入等待状态直到被通知(signal)或中断，当前线程进入后台运行状态且从await()方法返回
  * 其他线程调用该Condition的signal或者signalAll方法，而当前线程被选中唤醒
  * 1、其他线程（interrupt）中断当前线程
  * 2、如果当前等待线程从await方法返回，那么表明当前线程已经获取了Condition对象的锁
  */
 void await() throws InterruptedException;

 /**
  * 当前线程进入等待状态直到被通知，对中断不响应
  */
 void awaitUninterruptibly();

 /**
  * <pre> {@code
  * boolean aMethod(long timeout, TimeUnit unit) {
  * long nanos = unit.toNanos(timeout);
  * lock.lock();
  * try {
  *  while (!conditionBeingWaitedFor()) {
  *  if (nanos <= 0L)
  *   return false;
  *  nanos = theCondition.awaitNanos(nanos);
  *  }
  *  // ...
  * } finally {
  *  lock.unlock();
  * }
  * }}</pre>
  * 当前线程进入等待状态直到被通知、中断或超时。返回值表示剩余时间，如果在nanosTimeout纳秒之前被唤醒，那么返回值就是nanosTimeout-实际耗时
  * 返回值<=0说明超时
  * 
  */
 long awaitNanos(long nanosTimeout) throws InterruptedException;

 /**
  * 当前线程进入等待状态直到被通知、中断或超时，如果没有到指定时间被通知返回true，否则返回false
  */
 boolean await(long time, TimeUnit unit) throws InterruptedException;

 /**
  * 唤醒一个等待在Condition上的线程，该线程从等待方法返回之前必须获得与Condition相关联的锁
  */
 void signal();

public class BoundedQueue<T> {
 private Object[] items;

 // 添加的下标，删除的下标和数组当前数量
 private int addIndex,removeIndex,count;
 private Lock lock = new ReentrantLock();
 private Condition notEmpty = lock.newCondition();
 private Condition notFull = lock.newCondition();
 public BoundedQueue(int size){
  items = new Object[size];
 }

 /**
  * 添加一个元素，如果数组满，则添加线程进入等待状态，直到有"空位"
  * @author fuyuwei
  * 2017年5月21日 下午6:14:55
  * @param t
  * @throws InterruptedException
  */
 public void add(T t) throws InterruptedException{
  lock.lock();
  try{
   while(count == items.length){
    notFull.await();
   }
   items[addIndex] = t;
   if(++addIndex == items.length)
    addIndex = 0;
   ++count;
   notEmpty.signal();
  }finally{
   lock.unlock();
  }
 }

 /**
  * 由头部删除一个元素，如果数组空，则删除线程进入等待状态，直到有新添加元素
  * @author fuyuwei
  * 2017年5月21日 下午6:20:54
  * @return
  * @throws InterruptedException
  */
 @SuppressWarnings("unchecked")
 public T remove() throws InterruptedException{
  lock.lock();
  try{
   while(count == 0)
    notEmpty.await();
   Object x = items[removeIndex];
   if(++removeIndex == items.length)
    removeIndex = 0;
   --count;
   notFull.signal();
   return (T)x;
  }finally{
   lock.unlock();
  }
 }
}

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)
   unlinkCancelledWaiters();
  if (interruptMode != 0)
   reportInterruptAfterWait(interruptMode);
 }

public final void signal() {
  if (!isHeldExclusively())
   throw new IllegalMonitorStateException();
  Node first = firstWaiter;
  if (first != null)
   doSignal(first);
 }