生产者消费者的3种实现方式

叁歲伎倆 2022-05-20 02:16 131阅读 0赞

/** * 一、使用synchronized、wait、notify实现生产者，消费者模式 */
    public class ProducerConsumerWaitNofityAll { 
        public static void main(String[] args) {
            Queue<Integer> queue = new LinkedList<Integer>();
            new Thread(new Producer1(queue)).start();
            new Thread(new Consumer1(queue)).start();
        }
    }
    class Producer1 implements Runnable{
        private final Queue<Integer> queue;
        public Producer1(Queue<Integer> queue){
            this.queue = queue;
        }
        public void run() {
            while(true){
                synchronized (queue) {
                    while(queue.size() >= 10){
                        try {
                            System.out.println("queue is full,waiting...");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    int i = new Random().nextInt();
                    System.out.println("producer:"+i);
                    queue.add(i);
                    queue.notifyAll();
                }
            }
        }
    }
    class Consumer1 implements Runnable{
        private final Queue<Integer> queue;
        public Consumer1(Queue<Integer> queue){
            this.queue = queue;
        }
        public void run() {
            while(true){
                synchronized (queue) {
                    while(queue.size() == 0){
                        try {
                            System.out.println("queue is empty,waiting...");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    int i = queue.poll();
                    System.out.println("consumer:"+i);
                    queue.notifyAll();
                }
            }
        }
    
    }
    
    
    /** * 二、使用lock、condition、await、signal实现生产者，消费者模式 */
    public class ProducerConsumerAwaitSignalAll { 
        public static void main(String[] args) {
            Lock lock = new ReentrantLock();
            Condition prodCondition = lock.newCondition();
            Condition consCondition = lock.newCondition();
            Queue<Integer> queue = new LinkedList<Integer>();
            new Thread(new Producer2(lock, prodCondition, consCondition, queue)).start();
            new Thread(new Consumer2(lock, prodCondition, consCondition, queue)).start();
        }
    }
    
    class Producer2 implements Runnable{
        private Lock lock;
        private Condition prodCondition;
        private Condition consCondition;
        private Queue<Integer> queue;
        public Producer2(Lock lock,Condition prodCondition,Condition consCondition,Queue<Integer> queue){
            this.lock = lock;
            this.prodCondition = prodCondition;
            this.consCondition = consCondition;
            this.queue = queue;
        }
        public void run() {
            while(true){
                lock.lock();
                try {
                    while(queue.size() >= 10){
                        System.out.println("queue is full,waiting...");
                        prodCondition.await();
                    }
                    int i = new Random().nextInt();
                    System.out.println("producer:"+i);
                    queue.add(i);
                    consCondition.signalAll();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    lock.unlock();
                }
            }
        }
    }
    
    class Consumer2 implements Runnable{
        private Lock lock;
        private Condition prodCondition;
        private Condition consCondition;
        private Queue<Integer> queue;
        public Consumer2(Lock lock, Condition prodCondition,Condition consCondition, Queue<Integer> queue) {
            this.lock = lock;
            this.prodCondition = prodCondition;
            this.consCondition = consCondition;
            this.queue = queue;
        }
        public void run() {
            while(true){
                lock.lock();
                try {
                    while(queue.size() == 0){
                        System.out.println("queue is empty,waiting...");
                        consCondition.await();
                    }
                    int i = queue.poll();
                    System.out.println("consumer:"+i);
                    prodCondition.signalAll();
                } catch (Exception e) {
                    e.printStackTrace();
                }finally{
                    lock.unlock();
                }
            }
        }
    }
    
    
    /** * 三、使用BlockingQueue实现生产者，消费者模式。 * 经典的方法是使用wait和notify方法在生产者和消费者，BlockingQueue隐含的提供了这些控制，是线程安全的。 * 可以安全地与多个生产者和多个使用者一起使用 */
    public class ProducerConsumerBlockingQueue { 
        public static void main(String[] args) {
            BlockingQueue<Integer> sharedQueue = new LinkedBlockingQueue<Integer>();
            Thread prodThread = new Thread(new Producer(sharedQueue));
            Thread consThread1 = new Thread(new Consumer(sharedQueue));
            Thread consThread2 = new Thread(new Consumer(sharedQueue));
            prodThread.start();
            consThread1.start();
            consThread2.start();
        }
    }
    /** * 生产者 */
    class Producer implements Runnable{
        private final BlockingQueue<Integer> sharedQueue;
        public Producer(BlockingQueue<Integer> sharedQueue){
            this.sharedQueue = sharedQueue;
        }
        public void run() {
            for(int i= 0;i<10;i++){
                System.out.println("Producer: "+Thread.currentThread().getName()+" "+i);
                try {
                    sharedQueue.put(i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    /** * 消费者 */
    class Consumer implements Runnable{
        private final BlockingQueue<Integer> sharedQueue;
        public Consumer(BlockingQueue<Integer> sharedQueue){
            this.sharedQueue = sharedQueue;
        }
        public void run() {
            while(true){
                try {
                    System.out.println("Consumed: "+Thread.currentThread().getName()+" "+ sharedQueue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    }