ReentrantLock与Condition构造有界缓存队列与数据栈

Posted 2020-10-12 chen_yong

通过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 }