多线程应用——生产者消费者问题

Posted 2021-11-25 zhaodongge

前言

生产者和消费者问题是多线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。

在JAVA中实现生产者消费者问题时，有三种常用的方式：

• 使用Object的wait/notify的消息通知机制
• 使用Lock的Condition的await/signal的消息通知机制
• 使用BlockingQueue实现。本文主要将这三种实现方式进行总结归纳

下面本文将分别采用以上三种方式实现生产者消费者问题，诸如此类的多线程问题亦可如此实现。

1. 使用Object的wait/notify的消息通知机制

Java 中，可以通过配合调用 Object 对象的wait() 方法和 notify()方法或 notifyAll()方法来实现线程间的通信。在线程中调用 wait() 方法，将阻塞当前线程，直至等到其他线程调用了调用 notify()方法或 notifyAll()方法进行通知之后，当前线程才能从wait()方法出返回，继续执行下面的操作。下面详细介绍这三种方法。

• wait()方法
  该方法用来将当前线程置入休眠状态，直到接到通知或被中断为止。在调用 wait()之前，线程必须要获得该对象的对象监视器锁，即只能在同步方法或同步块中调用 wait()方法。调用wait()方法之后，当前线程会释放锁。如果调用wait()方法时，线程并未获取到锁的话，则会抛出IllegalMonitorStateException异常，这是以个RuntimeException。如果再次获取到锁的话，当前线程才能从wait()方法处成功返回。
  
  
• notify()方法
  该方法也要在同步方法或同步块中调用，即在调用前，线程也必须要获得该对象的对象级别锁，如果调用 notify()时没有持有适当的锁，也会抛出 IllegalMonitorStateException。该方法任意从等待状态的线程中挑选一个进行通知，使得调用wait()方法的线程从等待队列移入到同步队列中，等待有机会再一次获取到锁，从而使得调用wait()方法的线程能够从wait()方法处退出。调用notify后，当前线程不会马上释放该对象锁，要等 到程序退出同步块后，当前线程才会释放锁。
  
  
• notifyAll()
  该方法与 notify ()方法的工作方式相同，重要的一点差异是：notifyAll()使所有原来在该对象上等待的线程统统退出等待状态，使得他们全部从等待队列中移入到同步队列中去，等待下一次能够有机会获取到对象监视器锁。

在使用wait()和notify()实现生产者消费者问题时需要用到同步锁关键字synchronized，关于同步锁的概念请参考这里。

/*通过wait()和notifyall()消息机制实现生产者消费者问题*/
public class ProducerAndConsumerDemo2 {
    //缓冲区产品数量
    private static Integer count=0;
    //缓冲区大小
    private static final Integer BUFFER_SIZE=10;

    public static void main(String[] args) {
        ProducerAndConsumerDemo2 producerAndConsumerDemo2=new ProducerAndConsumerDemo2();
        for (int i = 0; i < 20; i++) {
            new Thread(producerAndConsumerDemo2.new Producer()).start();
            new Thread(producerAndConsumerDemo2.new Consumer()).start();
        }
    }

    class Producer implements Runnable{
        @Override
        public void run() {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (this){
                while(count==BUFFER_SIZE){
                    try {
                        this.wait();//缓冲区已满，本线程陷入阻塞
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //此时生产者线程由于消费者消费了物品跳出阻塞
                //生产一件物品，那么所以因为缺少物品而阻塞的消费者进程也应该解锁
                count++;
                System.out.println("生产者生产了一件物品，现在一共有"+count+"件物品！");
                this.notifyAll();
            }
        }
    }

    class Consumer implements Runnable{
        @Override
        public void run() {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (this){
                while(count==0){
                    try {
                        this.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                count--;
                System.out.println("消费者消费了一件物品，现在一共有"+count+"件物品！");
                this.notifyAll();
            }
        }
    }
}

2 可重入锁ReentrantLock的实现

java.util.concurrent.lock 中的 Lock 框架是锁定的一个抽象，通过对lock的lock()方法和unlock()方法实现了对锁的显示控制，而synchronize()则是对锁的隐性控制。

/*通过lock类的实例对象来实现生产者消费者问题*/
public class ProducerAndConsumerDemo1 {
    //缓冲区产品数量
    private static Integer count = 0;
    //缓冲区总大小
    private static final Integer BUFFER_SIZE =10;
    //锁定义
    Lock lock=new ReentrantLock();
    //定义缓冲区为满时的condition
    private final Condition FULL=lock.newCondition();
    //定义缓冲区为空时的condition
    private final Condition EMPTY=lock.newCondition();

    public static void main(String[] args) {
        ProducerAndConsumerDemo1 producerAndConsumerDemo1 = new ProducerAndConsumerDemo1();
        for (int i = 0; i <50 ; i++) {
            new Thread(producerAndConsumerDemo1.new Producer()).start();
            new Thread(producerAndConsumerDemo1.new Consumer()).start();
        }
    }

    class Producer implements Runnable{
        @Override
        public void run() {
            //System.out.println("进来消费者啦！");
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            lock.lock();
            try {
                while (count == BUFFER_SIZE) {
                    try {
                        FULL.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                count++;
                System.out.println("生产者生产了一件物品，现在一共有"+count+"件物品！");
                //唤醒沉睡的消费者
                EMPTY.signal();
            } finally {
                lock.unlock();
            }
        }
    }

    class Consumer implements Runnable{
        @Override
        public void run() {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            lock.lock();
            try{
                while(count==0){
                    try {
                        EMPTY.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                count--;
                System.out.println("消费者消费了一个物品，现在一共有"+count+"件物品！");
                //唤醒沉睡的生产者
                FULL.signal();
            }finally {
                lock.unlock();
            }
            }
    }
}

---

参考资料

https://baike.baidu.com/item/synchronized/8483356?fr=aladdin
https://www.cnblogs.com/chentingk/p/6497107.html

---