使用ReetrantLock和Candition实现生产者消费者模型

Bertha 。 2023-07-11 11:19 9阅读 0赞

package com.zjw;
    
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * 生产者消费者问题 */
    public class ReetrantLockTest { 
    
      static class MyContainer<T> { 
    
        private List<T> container = new LinkedList<>();
        private Integer maxSize;
        private Integer size = 0;
    
        private Lock lock = new ReentrantLock();
        private Condition producer = lock.newCondition();
        private Condition consumer = lock.newCondition();
    
        MyContainer(Integer maxSize) { 
          this.maxSize = maxSize;
        }
    
        void put(T t) throws InterruptedException { 
          lock.lock();
          // 当容器装满时让生产者等待，使用while循环检测
          while (this.size == this.maxSize) { 
            System.out.println("+++生产者等待...");
            producer.await(); // 会将锁释放出来，在被唤醒时需要重新获取锁
          }
          container.add(t);
          size++;
          System.out.println("+生产: " + t + ", current size: " + size);
          consumer.signalAll();
          lock.unlock();
        }
    
        T get() throws InterruptedException { 
          lock.lock();
          // 当容器为空时让消费者等待，使用while循环检测
          while (this.size == 0) { 
            System.out.println("---消费者等待...");
            consumer.await(); // 会将锁释放出来，在被唤醒时需要重新获取锁
          }
          T result = container.remove(0);
          size--;
          producer.signalAll();
          System.out.println("-消费: " + result + ", current size: " + size);
          lock.unlock();
          return result;
        }
      }
    
      public static void main(String[] args) throws InterruptedException { 
        MyContainer<Integer> container = new MyContainer<>(5);
        Random random = new Random();
        // 消费者 5个
        for (int i = 0; i < 5; i++) { 
          new Thread(() -> { 
            while (true) { 
              try { 
                TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));
                container.get();
              } catch (InterruptedException e) { 
                System.out.println("An error: " + e);
                e.printStackTrace();
              }
            }
          }).start();
        }
        TimeUnit.SECONDS.sleep(5);
        // 生产者 5个
        for (int i = 0; i < 5; i++) { 
          new Thread(() -> { 
            while (true) { 
              try { 
                TimeUnit.MILLISECONDS.sleep(random.nextInt(5000));
                container.put(random.nextInt(10000));
              } catch (InterruptedException e) { 
                System.out.println("An error: " + e);
                e.printStackTrace();
              }
            }
          }).start();
        }
      }
    }