用代码实现生产者消费者模型

野性酷女 2022-09-10 08:14 99阅读 0赞

题目：代码实现生产者消费者模型，3个生产者每隔3秒生产1个单位的数据，2个消费者每隔1秒消费一个单位的数据，共享资源是有界的，可以用数组或队列。要求不能出现死锁或者占用CPU不释放的情况。

方法一：用synchronized关键字的wait/notify模式加锁实现

// 共享资源
    public class Resource { 
        private LinkedList<String> commonList;
        private int capacity = 0;
    
        public Resource(int capacity) { 
            this.capacity = capacity;
            this.commonList = new LinkedList<>();
        }
        public void addLast(String item) { 
            commonList.addLast(item);
        }
        public void removeFirst() { 
            commonList.removeFirst();
        }
    
        public boolean isFull() { 
            return commonList.size() == this.capacity;
        }
    }
    // 生产者
    public class Producer extends Thread { 
    
        private int      duration;
        private Resource resource;
    
        public Producer (int duration, Resource resource) { 
            this.duration = duration;
            this.resource = resource;
        }
    
        @Override
        public void run() { 
            while (true) { 
                try { 
                    synchronized (resource) { 
                        if (!resource.isFull()) { 
                            resource.addLast("Producer_" + Thread.currentThread().getName());
                            // notify/notifyAll() 方法，唤醒一个或多个正处于等待状态的线程
                            resource.notifyAll();
                        } else { 
                            // wait()使当前线程阻塞，前提是必须先获得锁配合synchronized 关键字使用
                            resource.wait();
                        }
                    }
                    Thread.sleep(1000 * duration);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    // 消费者
    public class Consumer extends Thread { 
    
        private int      duration;
        private Resource resource;
    
        public Consumer (int duration, Resource resource) { 
            this.duration = duration;
            this.resource = resource;
        }
    
        @Override
        public void run() { 
            while (true) { 
                try { 
                    synchronized (resource) { 
                        // 共享区有数据才能取数据
                        if (resource.size() > 0) { 
                            resource.removeFirst();
                            // 消费数据后唤醒生产者
                            resource.notifyAll();
                        } else { 
                            // wait()使当前线程阻塞，前提是 必须先获得锁，配合synchronized 关键字使用
                            resource.wait();
                        }
                    }
                    Thread.sleep(1000 * duration);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    // 测试
    public class TestMain { 
        public static void main(String[] args) { 
            Resource resource = new Resource(10);
            new Producer(3, resource).start();
            new Producer(3, resource).start();
            new Producer(3, resource).start();
            new Consumer(1, resource).start();
            new Consumer(1, resource).start();
        }
    }

方法二：用阻塞队列来充当共享资源，好处是代码中不用再显式的处理锁，例如ArrayBlockingQueue内部就已经使用一个ReentrantLock和两个Condition来实现了对put操作与take操作的阻塞/唤醒。这样写的前提是你知道BlockingQueue解决了多线程环境下的数据共享。

![143badee75724ce0b442dc941318ebf5.png][]

// 生产者
    public class Producer extends Thread { 
        private int   duration;
        private BlockingQueue<String> blockingQueue;
    
        public Producer (int duration, BlockingQueue<String> blockingQueue) { 
            this.duration = duration;
            this.blockingQueue = blockingQueue;
        }
    
        @Override
        public void run() { 
            while (true) { 
                try { 
                    // 队列满时，put操作将被阻塞直到队列有空间为止
                    blockingQueue.put("Producer_" + Thread.currentThread().getName());
                    Thread.sleep(1000 * duration);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    // 消费者
    public class Consumer extends Thread { 
        private int                   duration;
        private BlockingQueue<String> blockingQueue;
    
        public Consumer (int duration, BlockingQueue<String> blockingQueue) { 
            this.duration = duration;
            this.blockingQueue = blockingQueue;
        }
    
        @Override
        public void run() { 
            while (true) { 
                try { 
                    String e = blockingQueue.take();
                    Thread.sleep(1000 * duration);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }
    
    
    public class TestMain { 
        public static void main(String[] args) { 
            BlockingQueue<String> resource = new ArrayBlockingQueue<>(24);
            new Producer(3, resource).start();
            new Producer(3, resource).start();
            new Producer(3, resource).start();
            new Consumer(1, resource).start();
            new Consumer(1, resource).start();
        }
    }

[143badee75724ce0b442dc941318ebf5.png]: /images/20220829/d60cda70ac1945ec9dc2a2596e47dba7.png