java的生产者消费者代码实现（使用wait()、notify()、notifyAll()）、阻塞队列实现

ゞ浴缸里的玫瑰 2024-03-16 18:15 9阅读 0赞

生产者消费者问题也是等待唤醒机制，是一个十分经典的多线程协作的模式。  
一、常见方法  
![在这里插入图片描述][1dfb7147b4644fbd8560b77165f608a6.png]  
例如，设生产者消费者问题中的缓冲池大小为1。首先利用操作系统中的信号量机制对问题进行分析:  
![在这里插入图片描述][995d80c1b0144c14ae15d68ddef52a96.png]  
![在这里插入图片描述][81d76dbb058f41a394eb8b6f51733b07.png]  
java代码实现：  
缓冲池：

public class Buffer {
        
        //缓冲池大小为1
    
     //标志缓冲池中是否有产品的变量，0表示无产品，1表示有产品
       public static int goodFlag = 0;
    
       //总次数,表示消费者生产产品的上限
        public static int count = 10;
    
        //锁对象，互斥地访问goodFlag变量
        public static Object lock = new Object();
    
    //同步方法，用于互斥地使用缓冲池，使用同步方法时阻塞和唤醒操作底层已经实现了
        public synchronized static void useBuffer(String name,Thread t){
        
            if("生产者".equals(name)){
        
                System.out.println(t.getName()+"生产了一个产品");
            }
            else if("消费者".equals(name))
            {
        
                System.out.println(t.getName()+"在消费产品，还能再消费"+Buffer.count+"个产品");
            }
        }
    
    }
    
    生产者：
    ```java
    public class Producer extends Thread{
        
        @Override
        public void run() {
        
             while(true){
        
                    synchronized (Buffer.lock){
        
                        if(Buffer.count==0){
        
                            break;
                        }
                        else{
        
                            if(Buffer.goodFlag==1){
        
                                try {
        
                                    //缓冲池内已有产品，生产者进程阻塞
                                    Buffer.lock.wait();
                                } catch (InterruptedException e) {
        
                                    e.printStackTrace();
                                }
                            }
                            else{
        
                                Buffer.useBuffer("生产者",Thread.currentThread());
                                Buffer.goodFlag=1;
                                Buffer.lock.notifyAll();//唤醒锁的阻塞队列上的线程
                            }
                        }
                    }
             }
        }
    }

消费者：

public class Consumer extends Thread{
        
        @Override
        public void run() {
        
            while(true) {
        
                synchronized (Buffer.lock){
        //Buffer.lock为锁
                    if(Buffer.count == 0)
                    {
        
                        //生产者已经生产了10个产品
                        break;
                    }
                    else{
        
                        if(Buffer.goodFlag == 0)
                        {
        
                            try {
        
                                Buffer.lock.wait();//使用锁对象使对象进入阻塞队列
                            } catch (InterruptedException e) {
        
                                e.printStackTrace();
                            }
                        }else{
        
                            Buffer.count--;
                            Buffer.useBuffer("消费者",Thread.currentThread());
                            Buffer.lock.notifyAll();//唤醒锁对象的阻塞队列中的所有线程
                            Buffer.goodFlag=0;
                        }
                    }
                }
            }
        }
    }

测试运行：
    
    ```java
    public class TestThreads {
        public static void main(String[] args) {
            Producer p1 = new Producer();
            Consumer c1 = new Consumer();
            Producer p2 = new Producer();
            Consumer c2 = new Consumer();
            c1.setName("消费者1");
            p1.setName("生产者1");
            c2.setName("消费者2");
            p2.setName("生产者2");
            c1.start();
            p1.start();
            c2.start();
            p2.start();
        }
    }

效果如下：  
![在这里插入图片描述][c4068654cf4649b1b57323e065a93e66.png]  
二、阻塞队列实现等待唤醒机制  
阻塞队列实现了下图中的红色框的四个接口：  
![在这里插入图片描述][97e7d6f3b2d44658b16edd592eb24f21.png]  
![在这里插入图片描述][28e93334b6ce46e8a0e2b74cc3e0eac2.png]

接口不能直接定义对象，需要用上图中两个蓝色框中的两个实现类定义阻塞队列的对象。  
阻塞队列相当于连接生产者和消费者之间的通道，相当于缓冲池，可以理解为缓冲区大小大于1，且是一个队列。  
使用阻塞队列简化上述代码。  
生产者：

public class Producer extends Thread{
        
        ArrayBlockingQueue<String> queue;
        public Producer(ArrayBlockingQueue<String> queue){
        
            this.queue=queue;
        }
        @Override
        public void run() {
        
             while(true){
        
                 try {
        
                     queue.put("产品");//将产品放到阻塞队列中
                     System.out.println(getName()+"生产了一个产品放到缓冲池中");
                 } catch (InterruptedException e) {
        
                     e.printStackTrace();
                 }
             }
        }
    }

消费者：

public class Consumer extends Thread{
        
        ArrayBlockingQueue<String> queue;
        public Consumer(ArrayBlockingQueue<String> queue){
        
            this.queue=queue;
        }
        @Override
        public void run() {
        
            while(true) {
        
                try {
        
                    String good = queue.take();//从阻塞队列中取产品
                    System.out.println(getName()+"消费了一个"+good);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

测试运行：

public class TestThreads {
        
        public static void main(String[] args) {
        
            ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
                    Producer producer1 = new Producer(queue);
            Producer producer2 = new Producer(queue);
            Consumer consumer1 = new Consumer(queue);
            Consumer consumer2 = new Consumer(queue);
            producer1.setName("生产者1");
            consumer1.setName("消费者1");
            producer2.setName("生产者2");
            consumer2.setName("消费者2");
            producer1.start();
            consumer1.start();
            producer2.start();
            consumer2.start();
        }
    }

结果：  
![在这里插入图片描述][c77b2308012d45a5a0ec212190b1ba53.png]  
可以看到，运行结果显示阻塞队列大小为1，生产者却可以连续生产，其实并不是这样。出现这种状况的原因如下图，生产者放产品和打印输出语句并不一定一次性执行完，由于多个线程并发执行导致这种错觉。  
![在这里插入图片描述][73f33970fab442248cadc14228bffd33.png]

[1dfb7147b4644fbd8560b77165f608a6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/4392dbb0248b4d508a520d040ee62c6c.png
[995d80c1b0144c14ae15d68ddef52a96.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/ae4c2be79f68481a9251b071196f0843.png
[81d76dbb058f41a394eb8b6f51733b07.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/ae44be453ed346aeb0d823c59fed2c95.png
[c4068654cf4649b1b57323e065a93e66.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2551393bac324b0fa04358db4352d7be.png
[97e7d6f3b2d44658b16edd592eb24f21.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/517b302f7f2941a483e79c50139af608.png
[28e93334b6ce46e8a0e2b74cc3e0eac2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/c126de2fa8d54c899f22c25ef72fbbf1.png
[c77b2308012d45a5a0ec212190b1ba53.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/f78ba27e5ad44d0090c036f2e67bedc3.png
[73f33970fab442248cadc14228bffd33.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/22f156fa50514d57a42b04a8151b07f8.png