Java生产者与消费者模式的简单写法

末蓝、 2022-06-07 01:10 152阅读 0赞

生产者消费者模式是研究多线程程序的经典问题之一，它描述是有一块缓冲区作为缓存队列/仓库，生产者可以将产品放入队列，消费者则可以从队列中取走产品。大多数的后台服务程序并发控制的基本原理都可以归纳为生产者消费者模式。

1、使用Synchronized（）、wait() 、 notify()、notifyAll()方法实现：

package proAndCsmModel01;
    
    import java.util.LinkedList;
    
    /**
     * 实现缓冲区
     * 
     */
    public class Resource01 {
        //最大缓冲区
        private final int MAX_SIZE = 10;
        //缓冲区队列
        LinkedList<datatype> list = new LinkedList<>();
    
        /**
         * 生产资料同步方法
         */
        public  synchronized void increaseData(){
                while (list.size() >= MAX_SIZE){
                    try {
                        System.out.println(Thread.currentThread().getId()+"资料仓库已满!");
                        wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                datatype d = new datatype();
                d.setData((int) (Math.random()*1000));
                list.add(d);
                System.out.println(Thread.currentThread().getId()+"生产:"+d.getData()+" 库存量："+list.size());
                notifyAll();
    
        }
    
        /**
         * 消费资料同步方法
         */
        public synchronized void decreaseData(){
                while (list.size() <= 0){
                    try {
                        System.out.println(Thread.currentThread().getId()+"资料仓库为空!");
                        wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                datatype d = list.poll();
                System.out.println(Thread.currentThread().getId()+"消费："+d.getData()+" 库存量："+list.size());
                notifyAll();
        }
    }
    
    package proAndCsmModel01;
    
    /**
     * 生产者：生产资料
     * 
     */
    public class producer01 implements Runnable {
        private Resource01 resource01;
        producer01(Resource01 resource01){
            this.resource01 = resource01;
        }
        @Override
        public void run() {
            while (true){
                try {
                    //随机休眠后在生产资料
                    Thread.sleep((long) (Math.random()*1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //调用同步方法进行生产资料
                resource01.increaseData();
            }
        }
    }
    package proAndCsmModel01;
    
    /**
     * 消费者：消费资料
     * 
     */
    public class consumer01 implements Runnable {
        private Resource01 resource01;
        consumer01(Resource01 resource01){
            this.resource01 = resource01;
        }
        @Override
        public void run() {
             while (true){
                 try {
                     //随机休眠后再消费
                     Thread.sleep((long) (Math.random()*1000));
                 } catch (InterruptedException e) {
                     e.printStackTrace();
                 }
                 //调用同步方法进行消费资料
                 resource01.decreaseData();
             }
        }
    }
    package proAndCsmModel01;
    
    /**
     * 基本数据类型
     * 
     */
    public class datatype {
        private int data;
    
        public void setData(int data) {
            this.data = data;
        }
    
        public int getData() {
            return data;
        }
    }
    package proAndCsmModel01;
    
    /**
     * 调用
     *
     */
    public class test {
        public static void main(String agrs[]){
            Resource01 resource01 = new Resource01();;
            System.out.println(Thread.currentThread().getName());
    
            new Thread(new producer01(resource01)).start();
            new Thread(new producer01(resource01)).start();
            new Thread(new producer01(resource01)).start();
            new Thread(new producer01(resource01)).start();
            new Thread(new producer01(resource01)).start();
            new Thread(new producer01(resource01)).start();
    
            new Thread(new consumer01(resource01)).start();
            new Thread(new consumer01(resource01)).start();
            new Thread(new consumer01(resource01)).start();
    
        }
    }

2、ReentrantLock、await() 、signal()、signalAll()方法实现：

package exampletest.proAndCsmModel01;
    
    import java.util.LinkedList;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 实现缓冲区
     * 
     */
    public class Resource02 {
        //最大缓冲区
        private final int MAX_SIZE = 10;
        //缓冲区队列
        LinkedList<datatype> list = new LinkedList<>();
        //同步锁和条件
        private Lock lock = new ReentrantLock();
        private Condition full_cd = lock.newCondition();
        private Condition empty_cd = lock.newCondition();
    
        /**
         * 生产资料同步方法
         */
        public void increaseData(){
            lock.lock();
            while (list.size() >= MAX_SIZE){
                try {
                    System.out.println(Thread.currentThread().getId()+"资料仓库已满!");
                    full_cd.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            datatype d = new datatype();
            d.setData((int) (Math.random()*1000));
            list.add(d);
            System.out.println(Thread.currentThread().getId()+"生产:"+d.getData()+" 库存量："+list.size());
            empty_cd.signalAll();
            lock.unlock();
        }
    
        /**
         * 消费资料同步方法
         */
        public void decreaseData(){
            lock.lock();
            while (list.size() <= 0){
                try {
                    System.out.println(Thread.currentThread().getId()+"资料仓库为空!");
                    empty_cd.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            datatype d = list.poll();
            System.out.println(Thread.currentThread().getId()+"消费："+d.getData()+" 库存量："+list.size());
            full_cd.signalAll();
            lock.unlock();
        }
    }

3、使用BlockingQueue阻塞队列方法实现

package exampletest.proAndCsmModel01;
    
    import java.util.LinkedList;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 实现缓冲区
     * 
     */
    public class Resource03 extends Resource{
        //最大缓冲区
        private final int MAX_SIZE = 10;
        //缓冲区队列
        LinkedBlockingDeque<datatype> list = new LinkedBlockingDeque<>(MAX_SIZE);
    
        /**
         * 生产资料同步方法
         */
        @Override
        public void increaseData(){
    
            if (list.size() >= MAX_SIZE){
                System.out.println(Thread.currentThread().getId()+"资料仓库已满!");
            }
    
            datatype d = new datatype();
            d.setData((int) (Math.random()*1000));
            try {
                list.put(d);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getId()+"生产:"+d.getData()+" 库存量："+list.size());
        }
    
        /**
         * 消费资料同步方法
         */
        @Override
        public void decreaseData(){
            if (list.size() <= 0){
                System.out.println(Thread.currentThread().getId()+"资料仓库为空!");
            }
            datatype d = null;
            try {
                d = list.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getId()+"消费："+d.getData()+" 库存量："+list.size());
    
        }
    }