public class BoundedQueue {
  /**
   * 生产者容器
   */
  private LinkedList<Object> buffer;
  /**
   * 容器最大值是多少
   */
  private int maxSize;
  /**
   * 锁
   */
  private Lock lock;
  /**
   * 满了
   */
  private Condition fullCondition;
  /**
   * 不满
   */
  private Condition notFullCondition;
  BoundedQueue(int maxSize) {
    this.maxSize = maxSize;
    buffer = new LinkedList<Object>();
    lock = new ReentrantLock();
    fullCondition = lock.newCondition();
    notFullCondition = lock.newCondition();
  }
  /**
   * 生产者
   *
   * @param obj
   * @throws InterruptedException
   */
  public void put(Object obj) throws InterruptedException {
    //获取锁
    lock.lock();
    try {
      while (maxSize == buffer.size()) {
        System.out.println(Thread.currentThread().getName() + "此时队列满了，添加的线程进入等待状态");
        // 队列满了，添加的线程进入等待状态
        notFullCondition.await();
      }
      buffer.add(obj);
      //通知
      fullCondition.signal();
    } finally {
      lock.unlock();
    }
  }
  /**
   * 消费者
   *
   * @return
   * @throws InterruptedException
   */
  public Object take() throws InterruptedException {
    Object obj;
    lock.lock();
    try {
      while (buffer.size() == 0) {
        System.out.println(Thread.currentThread().getName() + "此时队列空了线程进入等待状态");
        // 队列空了线程进入等待状态
        fullCondition.await();
      }
      obj = buffer.poll();
      //通知
      notFullCondition.signal();
    } finally {
      lock.unlock();
    }
    return obj;
  }
  public static void main(String[] args) {
    // 初始化最大能放2个元素的队列
    BoundedQueue boundedQueue = new BoundedQueue(2);
    for (int i = 0; i < 3; i++) {
      Thread thread = new Thread(() -> {
        try {
          boundedQueue.put("元素");
          System.out.println(Thread.currentThread().getName() + "生产了元素");
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      });
      thread.setName("线程" + i);
      thread.start();
    }
    for (int i = 0; i < 3; i++) {
      Thread thread = new Thread(() -> {
        try {
          boundedQueue.take();
          System.out.println(Thread.currentThread().getName() + "消费了元素");
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      });
      thread.setName("线程" + i);
      thread.start();
    }
    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

输出结果:

5. 源码分析

Condition接口中的方法

1. await()

       实现可中断条件等待,其实我们以上案例是利用ReentrantLock来实现的生产者消费者案例,进去看源码发现其实实现该方法的是 AbstractQueuedSynchronizer 中ConditionObject实现的
       将节点添加进同步队列中，并要么立即唤醒线程，要么等待前驱节点释放锁后将自己唤醒，无论怎样，被唤醒的线程要从哪里恢复执行呢？调用了await方法的地方

中断模式interruptMode这个变量记录中断事件，该变量有三个值：

1. 0 ： 代表整个过程中一直没有中断发生。
2. THROW_IE ： 表示退出await()方法时需要抛出InterruptedException，这种模式对应于中断发生在signal之前
3. REINTERRUPT ： 表示退出await()方法时只需要再自我中断以下，这种模式对应于中断发生在signal之后。

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加节点到条件队列中
    Node node = addConditionWaiter();
     // 释放当前线程所占用的锁，保存当前的锁状态
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 如果当前队列不在同步队列中，说明刚刚被await, 还没有人调用signal方法，
    // 则直接将当前线程挂起
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this); // 线程挂起的地方
         // 线程将在这里被挂起，停止运行
        // 能执行到这里说明要么是signal方法被调用了，要么是线程被中断了
        // 所以检查下线程被唤醒的原因，如果是因为中断被唤醒，则跳出while循环
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
   // 线程将在同步队列中利用进行acquireQueued方法进行“阻塞式”争锁，
   // 抢到锁就返回，抢不到锁就继续被挂起。因此，当await()方法返回时，
   // 必然是保证了当前线程已经持有了lock锁
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    // 如果尾节点被cancel了，则先遍历整个链表，清除所有被cancel的节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 将当前线程包装成Node扔进条件队列
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 如果当前节点为空值那么新创建的node节点就是第一个等待节点
    if (t == null)
        firstWaiter = node;
    // 如果当前节点不为空值那么新创建的node节点就加入到当前节点的尾部节点的下一个
    else
        t.nextWaiter = node;
    lastWaiter = node; // 尾部节点指向当前节点
    return node; // 返回新加入的节点
}

private void unlinkCancelledWaiters() {
    // 获取队列的头节点
    Node t = firstWaiter;
    Node trail = null;
    // 当前节点不为空 
    while (t != null) {
       // 获取下一个节点
        Node next = t.nextWaiter;
        // 如果当前节点不是条件节点
        if (t.waitStatus != Node.CONDITION) {
            // 在队列中取消当前节点
            t.nextWaiter = null;
            if (trail == null)
                // 队列的头节点是当前节点的下一个节点
                firstWaiter = next;
            else
                // trail的 nextWaiter 指向当前节点t的下一个节点
                // 因为此时t节点已经被取消了
                trail.nextWaiter = next;
                // 如果t节点的下一个节点为空那么lastWaiter指向trail
            if (next == null)
                lastWaiter = trail;
        }
        else
            // 如果是条件节点 trail 指向当前节点
            trail = t;
        // 循环赋值遍历
        t = next;
    }
}

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        // 如果释放成功
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            // 节点的状态被设置成取消状态，从同步队列中移除
            node.waitStatus = Node.CANCELLED;
    }
}
public final boolean release(int arg) {
    // 尝试获取锁，如果获取成功，唤醒后续线程
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

 // 检查是否有中断，如果在发出信号之前被中断，则返回THROW_IE；
 // 在发出信号之后，则返回REINTERRUPT；如果没有被中断，则返回0。
private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        enq(node);
        return true;
    }
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}

   // ....
   while (!isOnSyncQueue(node)) {
        LockSupport.park(this); // 线程挂起的地方
         // 线程将在这里被挂起，停止运行
        // 能执行到这里说明要么是signal方法被调用了，要么是线程被中断了
        // 所以检查下线程被唤醒的原因，如果是因为中断被唤醒，则跳出while循环
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
   // 线程将在同步队列中利用进行acquireQueued方法进行“阻塞式”争锁，
   // 抢到锁就返回，抢不到锁就继续被挂起。因此，当await()方法返回时，
   // 必然是保证了当前线程已经持有了lock锁
   
   // 我们这里假设它获取到了锁了,由于我们这时
   // 的interruptMode = THROW_IE，则会跳过if语句。
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    // 遍历链表了，把链表中所有没有在等待的节点都拿出去，所以这里调用
    // unlinkCancelledWaiters方法，该方法我们在前面await()第一部分的分析
    // 的时候已经讲过了，它就是简单的遍历链表，找到所有waitStatus
    // 不为CONDITION的节点，并把它们从队列中移除
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    // 这里我们的interruptMode=THROW_IE，说明发生了中断，
    // 则将调用reportInterruptAfterWait
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
    }
 }
// 在interruptMode=THROW_IE时，就是简单的抛出了一个InterruptedException
private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}

final boolean transferAfterCancelledWait(Node node) {
// 线程A执行到这里，CAS操作将会失败
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) { 
        enq(node);
        return true;
    }
// 由于中断发生前，线程已经被signal了，则这里只需要等待线程成功进入同步即可
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}

    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    // 这里我们的interruptMode=THROW_IE，说明发生了中断，
    // 则将调用reportInterruptAfterWait
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
    }
 }
// 在interruptMode=THROW_IE时，就是简单的抛出了一个InterruptedException
private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
   // 这里并没有抛出中断异常，而只是将当前线程再中断一次。
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}

public final void awaitUninterruptibly() {
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    boolean interrupted = false;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if (Thread.interrupted())
       //  发生了中断后线程依旧留在了条件队列中，将会再次被挂起
            interrupted = true;
    }
    if (acquireQueued(node, savedState) || interrupted)
        selfInterrupt();
}

public final long awaitNanos(long nanosTimeout)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    final long deadline = System.nanoTime() + nanosTimeout;
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        if (nanosTimeout <= 0L) {
            transferAfterCancelledWait(node);
            break;
        }
        if (nanosTimeout >= spinForTimeoutThreshold)
            LockSupport.parkNanos(this, nanosTimeout);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
        nanosTimeout = deadline - System.nanoTime();
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null)
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
    return deadline - System.nanoTime();
}

public final boolean await(long time, TimeUnit unit)
        throws InterruptedException {
    long nanosTimeout = unit.toNanos(time);
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    final long deadline = System.nanoTime() + nanosTimeout;
    boolean timedout = false;
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        if (nanosTimeout <= 0L) {
            timedout = transferAfterCancelledWait(node);
            break;
        }
        if (nanosTimeout >= spinForTimeoutThreshold)
            LockSupport.parkNanos(this, nanosTimeout);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
        nanosTimeout = deadline - System.nanoTime();
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null)
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
    return !timedout;
}

public final boolean awaitUntil(Date deadline)
        throws InterruptedException {
    long abstime = deadline.getTime();
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    boolean timedout = false;
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        if (System.currentTimeMillis() > abstime) {
            timedout = transferAfterCancelledWait(node);
            break;
        }
        LockSupport.parkUntil(this, abstime);
        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);
    return !timedout;
}

public final void signal() {
 // getExclusiveOwnerThread() == Thread.currentThread(); 当前线
 // 程是不是独占线程
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 获取第一个阻塞线程节点
    Node first = firstWaiter;
    // 条件队列是否为空
    if (first != null)
        doSignal(first);
}
// 遍历整个条件队列，找到第一个没有被cancelled的节点，并将它添加到条件队列的末尾
// 如果条件队列里面已经没有节点了，则将条件队列清空
private void doSignal(Node first) {
    do {
        // 将firstWaiter指向条件队列队头的下一个节点
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 将条件队列原来的队头从条件队列中断开，则此时该节点成为一个孤立的节点
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

public final void signalAll() {
 // getExclusiveOwnerThread() == Thread.currentThread(); 当前线
 // 程是不是独占线程
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 获取第一个阻塞线程节点
    Node first = firstWaiter;
   // 条件队列是否为空
    if (first != null)
        doSignalAll(first);
}
// 移除并转移所有节点
private void doSignalAll(Node first) {
    // 清空队列中所有数据
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}
// 将条件队列中的节点一个一个的遍历到同步队列中
final boolean transferForSignal(Node node) {
  // 如果该节点在调用signal方法前已经被取消了，则直接跳过这个节点
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
    return false;
 // 利用enq方法将该节点添加至同步队列的尾部   
    Node p = enq(node); 
    // 返回的是前驱节点，将其设置SIGNAL之后，才会挂起
    // 当前节点
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}