通过ReentrantLock与Condition的设计，以数组为基础，可以实现简单的队列和栈的数据结构，临界阻塞的效果。
ReentrantLock相对于synchronized比较大的一个区别是有条件变量:Condition,很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。Condition（也称为条件队列 或条件变量）为线程提供了一个含义，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”）。因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是：以原子方式 释放相关的锁，并挂起当前线程，就像 Object.wait 做的那样。多个Condition需要绑定到同一锁上,可以实现队列与栈。
队列：先进先出的原则
栈：先进后出的原则
类一：模拟队列的读写操作
package reentranlock; import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; public class BoundedBufferQueue { static Lock lock = new ReentrantLock();
static Condition read = lock.newCondition();
static Condition write = lock.newCondition();
static Object [] data = new Object [10];// 构造一个缓存队列 private static int count = 0;// 用来标识队列中存放的数据量
private static int readIndex = 0;// 标识读取的下标
private static int writeIndex = 0;// 标识写入的下标 public static void put(Integer num) throws InterruptedException {
try {
lock.lock();
if (count == 10) {
write.await();// 数据量满了则阻塞写的操作
}
data[writeIndex] = num;
count++;
if (++writeIndex == 10) {// 循环写入数据
writeIndex = 0;
}
read.signal();// 触发读操作
} finally {
lock.unlock();
}
} public static Object take() throws InterruptedException {
Object result = null;
try {
lock.lock();
if (count == 0) {// 如果队列无数据量则阻塞读操作
read.await();
}
result = (Integer) data[readIndex];
count--;
if (++readIndex == 10) {// 循环取数据
readIndex = 0;
}
write.signal();// 触发写操作
} finally {
lock.unlock();
}
return result;
} // 下面是模拟读写操作过程,可以通过操作时间不同来验证队列读取。
public static void main(String[] args) throws InterruptedException { Runnable readThread = new Runnable() {
@Override
public void run() {
while(true){
for(int i=1;i<Integer.MAX_VALUE;i++){
try {
Integer o = (Integer) take();
System.out.println("读取："+o);
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} }
}; Runnable writeThread = new Runnable() {
@Override
public void run() {
while(true){
for(int i=1;i<Integer.MAX_VALUE;i++){
try {
put(i);
System.out.println("写入："+i);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} }
}; Thread read = new Thread(readThread);
Thread write = new Thread(writeThread); read.start();
Thread.currentThread().join(1000);
write.start();
} }
类二：模拟栈的读写操作
package reentranlock; import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; public class BoundedBufferStack { static Lock lock = new ReentrantLock();
static Condition read = lock.newCondition();
static Condition write = lock.newCondition();
static Object [] data = new Object [10];// 构造一个缓存栈 private static int count = 0;// 用来标识栈中存放的数据量
private static int index = 0;// 标识的下标 public static void put(Integer num) throws InterruptedException {
try {
lock.lock();
if (count == 10) {// 数据量满了则阻塞写操作
write.await();
}
data[index] = num;
count++;
index++;
if (index == 10) {
index = 0;
}
read.signal();// 触发读操作
} finally {
lock.unlock();
}
} public static Object take() throws InterruptedException {
Object result = null;
try {
lock.lock();
if (count == 0) {// 数据量为空则阻塞读操作
read.await();
}
if(index == 0 && count == 10){// 为了仿造栈的后进先出的模式,取最后写入的数据
index = 9;
}else{
index --;
}
result = (Integer) data[index];
count--;
if (index == 0) {
index = 0;
}
write.signal();// 触发写操作
} finally {
lock.unlock();
}
return result;
} // 下面是模拟读写操作过程,可以通过操作时间不同来验证栈的读取。
public static void main(String[] args) throws InterruptedException { Runnable readThread = new Runnable() {
@Override
public void run() {
while(true){
for(int i=1;i<Integer.MAX_VALUE;i++){
try {
Integer o = (Integer) take();
System.out.println("读取："+o);
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} }
}; Runnable writeThread = new Runnable() {
@Override
public void run() {
while(true){
for(int i=1;i<Integer.MAX_VALUE;i++){
try {
put(i);
System.out.println("写入："+i);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} }
}; Thread read = new Thread(readThread);
Thread write = new Thread(writeThread); write.start();
Thread.currentThread().join(1000);
read.start();
} }
ArrayBlockingQueue也是这种设计 "通过平衡生产者和消费者的处理能力来提高整体处理数据的速度"，只不过运用ArrayBlockingQueue不要担心非单一生产者/消费者场景下的系统假死问题，缓冲区空、缓冲区满的场景BlockingQueue都是定义了不同的Condition，所以不会唤醒自己的同类。