java ReentrantLock结合条件队列 实现生产者-消费者模式

 1 package reentrantlock;
 2 
 3 import java.util.ArrayList;
 4 
 5 public class ProviderAndConsumerTest {
 6 
 7     static ProviderAndConsumer providerAndConsumer  =  new ProviderAndConsumer();
 8 
 9     public static void main(String[] args) throws InterruptedException {
10 
11 //        new Thread(new GetRunnable(), "消费者002").start();
12 //        Thread.sleep(1000);
13 //        new Thread(new PutRunnable(), "生产者001").start();
14 
15         ArrayList<Thread> provider = new ArrayList<>();
16         for (int i = 0; i < 3; i++){
17             provider.add(new Thread(new PutRunnable(), "生产者00"+ (i+1)));
18         }
19 
20         ArrayList<Thread> consumer = new ArrayList<>();
21         for (int i = 0; i < 3; i++){
22             consumer.add(new Thread(new GetRunnable(), "      消费者--99"+ (i+1)));
23         }
24 
25 
26         for (Thread i :
27                 consumer) {
28             i.start();
29         }
30 
31         // 先让消费者线程全部饥饿，进入消费者条件队列中
32         Thread.sleep(10000);
33 
34         for (Thread i :
35                 provider) {
36             i.start();
37         }
38 
39 
40 
41 
42     }
43 
44     static class PutRunnable implements Runnable {
45 
46 
47         @Override
48         public void run() {
49             for (int i = 0; i < 3; i++){
50                 providerAndConsumer.put("   (" +Thread.currentThread().getName() + "_data_" + i + ")");
51                 try {
52                     // 调整睡眠时间，等同于调整生产者生产数据的频率，但是不准，因为跟生产者内部逻辑执行时间有很大关系
53                     Thread.sleep(500,1);
54                 } catch (InterruptedException e) {
55                     e.printStackTrace();
56                 }
57             }
58         }
59     }
60 
61     static class GetRunnable implements Runnable{
62 
63         @Override
64         public void run() {
65             while(true){
66                 providerAndConsumer.get();
67                 try {
68                     // 调整睡眠时间，等同于调整消费者消费数据的频率，但是不准，因为跟消费者内部执行时间有很大关系
69                     Thread.sleep(10,1);
70                 } catch (InterruptedException e) {
71                     e.printStackTrace();
72                 }
73             }
74         }
75     }
76 }

  1 package reentrantlock;
  2 
  3 import java.util.concurrent.locks.Condition;
  4 import java.util.concurrent.locks.ReentrantLock;
  5 
  6 public class ProviderAndConsumer {
  7 
  8     ReentrantLock reentrantLock = new ReentrantLock();
  9     Condition notEmpty = reentrantLock.newCondition();
 10     Condition notFull = reentrantLock.newCondition();
 11     int maxSize = 3;
 12     int putIndex = 0;
 13     int getIndex = 0;
 14     int realDataCount = 0;
 15     Object[] queue = new Object[maxSize];
 16 
 17     boolean isConsumerWait = false;
 18     boolean isProviderWait = false;
 19 
 20 
 21     public void put(Object data){
 22         System.out.println(Thread.currentThread().getName() + "线程，-----尝试加锁-----");
 23         reentrantLock.lock();
 24         System.out.println(Thread.currentThread().getName() + "线程，-----加锁成功-----");
 25         try {
 26 
 27             while (realDataCount == queue.length){
 28                 System.out.println(Thread.currentThread().getName() + "线程，-----数据满了，只好等待----" +
 29                         "哈哈要进生产者条件队列啦");
 30                 isProviderWait = true;
 31                 notFull.await();
 32             }
 33 
 34             queue[putIndex] = data;
 35             realDataCount++;
 36 
 37             putIndex++;
 38             if (putIndex == queue.length) putIndex = 0;
 39             System.out.println(Thread.currentThread().getName() + "线程，成功生产一个数据=" + data.toString());
 40 
 41             if (isConsumerWait){
 42                 System.out.println(Thread.currentThread().getName() + "线程，" +
 43                         " 未 激活消费者条件队列节点前，获取消费者队列长度"
 44                         + reentrantLock.getWaitQueueLength(notEmpty));
 45             }
 46             notEmpty.signal();
 47             if (isConsumerWait){
 48                 int length = reentrantLock.getWaitQueueLength(notEmpty);
 49                 if (length == 0){
 50                     isConsumerWait = false;
 51                 }
 52                 System.out.println(Thread.currentThread().getName() + "线程，" +
 53                         " 已 激活消费者条件队列节点后，获取消费者队列长度"
 54                         + reentrantLock.getWaitQueueLength(notEmpty));
 55             }
 56             // 调整睡眠时间，等同于调整生产者持有锁的时间
 57             Thread.sleep(1000);
 58         } catch (InterruptedException e) {
 59             e.printStackTrace();
 60         } finally {
 61             System.out.println(Thread.currentThread().getName() + "线程，成功解锁");
 62             reentrantLock.unlock();
 63         }
 64 
 65     }
 66 
 67 
 68     public void get(){
 69         System.out.println("\t\t" + Thread.currentThread().getName() + "线程，尝试加锁");
 70         reentrantLock.lock();
 71         System.out.println("\t\t" + Thread.currentThread().getName() + "线程，-----加锁成功----");
 72 
 73         try {
 74 
 75             while (realDataCount == 0){
 76                 System.out.println("\t\t" + Thread.currentThread().getName() + "线程，-----没有数据，只好等待----"+
 77                         "哈哈 要进消费者条件队列啦");
 78                 isConsumerWait = true;
 79                 notEmpty.await();
 80             }
 81 
 82             System.out.println("\t\t" + Thread.currentThread().getName() + "线程，成功消费一个数据=" + queue[getIndex]);
 83             realDataCount--;
 84 
 85             getIndex++;
 86             if (getIndex == queue.length) getIndex = 0;
 87 
 88             if (isProviderWait){
 89                 System.out.println("\t\t" + Thread.currentThread().getName() + "线程，" +
 90                         " 未 激活生产者条件队列节点前，获取生产者队列长度"
 91                         + reentrantLock.getWaitQueueLength(notFull));
 92             }
 93             notFull.signal();
 94             if (isProviderWait){
 95                 int length = reentrantLock.getWaitQueueLength(notFull);
 96                 if (length == 0){
 97                     isProviderWait = false;
 98                 }
 99                 System.out.println("\t\t" + Thread.currentThread().getName() + "线程，" +
100                         " 已 激活生产者条件队列节点后，获取生产者队列长度"
101                         + length);
102             }
103 
104         } catch (InterruptedException e) {
105             e.printStackTrace();
106         } finally {
107             System.out.println("\t\t" + Thread.currentThread().getName() + "线程，成功解锁");
108             reentrantLock.unlock();
109         }
110     }
111 
112 
113 
114 
115 
116 
117 }