As an example, suppose we have a bounded buffer which supports put and take methods. If a take is attempted on an empty buffer, then the thread will block until an item becomes available; if a put is attempted on a full buffer, then the thread will block until a space becomes available. We would like to keep waiting put threads and take threads in separate wait-sets so that we can use the optimization of only notifying a single thread at a time when items or spaces become available in the buffer. This can be achieved using two Condition instances.
class BoundedBuffer { final Lock lock = new ReentrantLock(); final Condition notFull = lock.newCondition(); final Condition notEmpty = lock.newCondition(); final Object[] items = new Object[100]; int putptr, takeptr, count; public void put(Object x) throws InterruptedException { lock.lock(); try { while (count == items.length) notFull.await(); items[putptr] = x; if (++putptr == items.length) putptr = 0; ++count; notEmpty.signal(); } finally { lock.unlock(); } } public Object take() throws InterruptedException { lock.lock(); try { while (count == 0) notEmpty.await(); Object x = items[takeptr]; if (++takeptr == items.length) takeptr = 0; --count; notFull.signal(); return x; } finally { lock.unlock(); } } } (The ArrayBlockingQueue class provides this functionality, so there is no reason to implement this sample usage class.) package condtion; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class ArrayBlockingQueueTest { //ArrayBlockingQueue 和 BouderyBuffer 的循环数组使用同样的容量 static final int Capacity = 5; public static void main(String[] args ) { // ArrayBlockingQueue的主要框架就是BoudedBuffer。 ArrayBlockingQueue blockQueue1 = new ArrayBlockingQueue<>( Capacity ); // 看看BoundedBuffer的执行效果 BoundedBuffer blockQueue2 = new BoundedBuffer(); // 在这个例子，入参可以替换，只是循环数组的容量不一样，效果有所不同 // blockQueue1/blockQueue2 Runnable taskPut = new RunnablePut( blockQueue2 ); Runnable taskTake = new RunnableTake( blockQueue2 ); new Thread( taskPut ).start(); new Thread( taskTake ).start(); } static class RunnableTake<T> implements Runnable { T queue ; public RunnableTake(T blockQueue2 ) { queue = blockQueue2 ; } @Override public void run() { for ( int i = 0; i < 100; i ++) { try { int result = 0; if ( queue instanceof BoundedBuffer) { result = ( int ) ((BoundedBuffer) queue ).take(); } else { result = ( int ) ((ArrayBlockingQueue) queue ).take(); } System. out .printf( "take result = %d\r\n" , result ); } catch (InterruptedException e ) { // TODO Auto-generated catch block e .printStackTrace(); } } } } static class RunnablePut<T> implements Runnable { T queue ; public RunnablePut(T blockQueue2 ) { queue = blockQueue2 ; } @Override public void run() { for ( int i = 0; i < 100; i ++) { try { System. out .printf( "goint to put : %d\r\n" , i ); if ( queue instanceof BoundedBuffer) { ((BoundedBuffer) queue ).put( i ); } else { ((ArrayBlockingQueue) queue ).put( i ); } } catch (InterruptedException e ) { // TODO Auto-generated catch block e .printStackTrace(); } } } } static class BoundedBuffer { final Lock lock = new ReentrantLock(); final Condition notFull = lock .newCondition(); final Condition notEmpty = lock .newCondition(); final Object[] items = new Object[ Capacity ]; //ptr作用：读写数组的数据。疑问：如何确保数据不被重复读取？ int putptr , takeptr ; /* count作用： * 前提说明： items是一个循环数组，默认容量是5，实例从Object[0] - Object[4]。 * 从而限定了put和take的操作的次数差值只能为[0-5]，避免xxxptr越界操作。 * * count == 0. 说明take和put的操作次数相等 * count == 5. 说明put的次数等于5。这是循环数组items[5]的极限. * count > 5 . 说明put的操作次数比take次数超出5次，而循环数组大小是5，说明有数据还没来得及被take就被覆盖了。 * count < 0 . 说明take操作比put还多，表示take获取到了还没put的数据。 * 因为>5和<0的情况是不允许的，因此count的边界是0-5.在这两个值出现的时候，进行Condition条件等待。 */ int count ; public void put(Object x ) throws InterruptedException { lock .lock(); try { while ( count == items . length ) { System. out .println( "Queue is Fullded" ); notFull .await(); } items [ putptr ] = x ; if (++ putptr == items . length ) { putptr = 0; } ++ count ; notEmpty .signal(); } finally { lock .unlock(); } } public Object take() throws InterruptedException { lock .lock(); try { while ( count == 0) { System. out .println( "Queue is Empty" ); notEmpty .await(); } Object x = items [ takeptr ]; if (++ takeptr == items . length ) { takeptr = 0; } -- count ; notFull .signal(); return x ; } finally { lock .unlock(); } } } }