通过ReentrantLock与Condition的设计，以数组为基础，可以实现简单的队列和栈的数据结构，临界阻塞的效果。
ReentrantLock相对于synchronized比较大的一个区别是有条件变量:Condition,很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。Condition（也称为条件队列 或条件变量）为线程提供了一个含义，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”）。因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是：以原子方式 释放相关的锁，并挂起当前线程，就像 Object.wait 做的那样。多个Condition需要绑定到同一锁上,可以实现队列与栈。
队列：先进先出的原则
栈：先进后出的原则
类一：模拟队列的读写操作
1 package reentranlock; 2 3 import java.util.concurrent.locks.Condition; 4 import java.util.concurrent.locks.Lock; 5 import java.util.concurrent.locks.ReentrantLock; 6 7 public class BoundedBufferQueue { 8 9 static Lock lock = new ReentrantLock(); 10 static Condition read = lock.newCondition(); 11 static Condition write = lock.newCondition(); 12 static Object [] data = new Object [10];// 构造一个缓存队列 13 14 private static int count = 0;// 用来标识队列中存放的数据量 15 private static int readIndex = 0;// 标识读取的下标 16 private static int writeIndex = 0;// 标识写入的下标 17 18 public static void put(Integer num) throws InterruptedException { 19 try { 20 lock.lock(); 21 if (count == 10) { 22 write.await();// 数据量满了则阻塞写的操作 23 } 24 data[writeIndex] = num; 25 count++; 26 if (++writeIndex == 10) {// 循环写入数据 27 writeIndex = 0; 28 } 29 read.signal();// 触发读操作 30 } finally { 31 lock.unlock(); 32 } 33 } 34 35 public static Object take() throws InterruptedException { 36 Object result = null; 37 try { 38 lock.lock(); 39 if (count == 0) {// 如果队列无数据量则阻塞读操作 40 read.await(); 41 } 42 result = (Integer) data[readIndex]; 43 count--; 44 if (++readIndex == 10) {// 循环取数据 45 readIndex = 0; 46 } 47 write.signal();// 触发写操作 48 } finally { 49 lock.unlock(); 50 } 51 return result; 52 } 53 54 // 下面是模拟读写操作过程,可以通过操作时间不同来验证队列读取。 55 public static void main(String[] args) throws InterruptedException { 56 57 Runnable readThread = new Runnable() { 58 @Override 59 public void run() { 60 while(true){ 61 for(int i=1;i<Integer.MAX_VALUE;i++){ 62 try { 63 Integer o = (Integer) take(); 64 System.out.println("读取："+o); 65 Thread.sleep(3000); 66 } catch (InterruptedException e) { 67 e.printStackTrace(); 68 } 69 } 70 } 71 72 } 73 }; 74 75 Runnable writeThread = new Runnable() { 76 @Override 77 public void run() { 78 while(true){ 79 for(int i=1;i<Integer.MAX_VALUE;i++){ 80 try { 81 put(i); 82 System.out.println("写入："+i); 83 Thread.sleep(1000); 84 } catch (InterruptedException e) { 85 e.printStackTrace(); 86 } 87 } 88 } 89 90 } 91 }; 92 93 Thread read = new Thread(readThread); 94 Thread write = new Thread(writeThread); 95 96 read.start(); 97 Thread.currentThread().join(1000); 98 write.start(); 99 } 100 101 }
类二：模拟栈的读写操作
1 package reentranlock; 2 3 import java.util.concurrent.locks.Condition; 4 import java.util.concurrent.locks.Lock; 5 import java.util.concurrent.locks.ReentrantLock; 6 7 public class BoundedBufferStack { 8 9 static Lock lock = new ReentrantLock(); 10 static Condition read = lock.newCondition(); 11 static Condition write = lock.newCondition(); 12 static Object [] data = new Object [10];// 构造一个缓存栈 13 14 private static int count = 0;// 用来标识栈中存放的数据量 15 private static int index = 0;// 标识的下标 16 17 public static void put(Integer num) throws InterruptedException { 18 try { 19 lock.lock(); 20 if (count == 10) {// 数据量满了则阻塞写操作 21 write.await(); 22 } 23 data[index] = num; 24 count++; 25 index++; 26 if (index == 10) { 27 index = 0; 28 } 29 read.signal();// 触发读操作 30 } finally { 31 lock.unlock(); 32 } 33 } 34 35 public static Object take() throws InterruptedException { 36 Object result = null; 37 try { 38 lock.lock(); 39 if (count == 0) {// 数据量为空则阻塞读操作 40 read.await(); 41 } 42 if(index == 0 && count == 10){// 为了仿造栈的后进先出的模式,取最后写入的数据 43 index = 9; 44 }else{ 45 index --; 46 } 47 result = (Integer) data[index]; 48 count--; 49 if (index == 0) { 50 index = 0; 51 } 52 write.signal();// 触发写操作 53 } finally { 54 lock.unlock(); 55 } 56 return result; 57 } 58 59 // 下面是模拟读写操作过程,可以通过操作时间不同来验证栈的读取。 60 public static void main(String[] args) throws InterruptedException { 61 62 Runnable readThread = new Runnable() { 63 @Override 64 public void run() { 65 while(true){ 66 for(int i=1;i<Integer.MAX_VALUE;i++){ 67 try { 68 Integer o = (Integer) take(); 69 System.out.println("读取："+o); 70 Thread.sleep(5000); 71 } catch (InterruptedException e) { 72 e.printStackTrace(); 73 } 74 } 75 } 76 77 } 78 }; 79 80 Runnable writeThread = new Runnable() { 81 @Override 82 public void run() { 83 while(true){ 84 for(int i=1;i<Integer.MAX_VALUE;i++){ 85 try { 86 put(i); 87 System.out.println("写入："+i); 88 Thread.sleep(1000); 89 } catch (InterruptedException e) { 90 e.printStackTrace(); 91 } 92 } 93 } 94 95 } 96 }; 97 98 Thread read = new Thread(readThread); 99 Thread write = new Thread(writeThread); 100 101 write.start(); 102 Thread.currentThread().join(1000); 103 read.start(); 104 } 105 106 }
ArrayBlockingQueue也是这种设计 "通过平衡生产者和消费者的处理能力来提高整体处理数据的速度"，只不过运用ArrayBlockingQueue不要担心非单一生产者/消费者场景下的系统假死问题，缓冲区空、缓冲区满的场景BlockingQueue都是定义了不同的Condition，所以不会唤醒自己的同类。