多线程-生产者和消费者模式的四种实现

小灰灰 2022-12-08 05:38 104阅读 0赞

## 什么是生产者和消费者模式： ##

生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此并不直接通信，而是通过阻塞队列进行通信，所以生产者生产完数据后不用等待消费者进行处理，而是直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列中获取数据，阻塞队列就相当于一个缓冲区，平衡生产者和消费者的处理能力。

## wait/notify和synchronized配合实现： ##

生产者和消费者线程各一条：

代码实现：

package ThreadDemo.ThreadExercise;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    
    /*
     * 生产者线程
     */
    class ProducerDemo implements Runnable{
        private List list;
        public ProducerDemo(List list){
            this.list=list;
        }
        @Override
        public void run() {
            while(true){
                Random random=new Random();
                synchronized(list){
                    if(list.size()>0){ //表明集合中有元素，此线程等待
                    //可以是while
                        try {
                            list.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    list.add(random.nextInt(100));//0-99的随机数；
                    System.out.println(Thread.currentThread().getName()+"  "+list.get(0));
                    list.notify(); //通知消费者，集合中已有元素。
                }
            }
        }
    }
    /*
     *  消费者线程
     */
    class ConsumerDemo implements Runnable{
        private List list;
        public ConsumerDemo(List list){
             this.list=list;
        }
        @Override
        public void run() {
            while(true){
                synchronized (list){
                    if(list.size()<1){//可以是while
                        try {
                            list.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println(Thread.currentThread().getName()+"  "+list.remove(0));
                    list.notify();
                }
            }
        }
    }
    public class ProducerAndConsumer {
        public static void main(String[] args) {
            List list=new ArrayList();
            Thread thread1=new Thread(new ProducerDemo(list));
            thread1.setName("生产者线程_");
            Thread thread2=new Thread(new ConsumerDemo(list));
            thread2.setName("......消费者线程_");
            thread2.start();
            thread1.start();
        }
    }

测试结果：

生产者线程_  88
    ......消费者线程_  88
    生产者线程_  77
    ......消费者线程_  77
    生产者线程_  49
    ......消费者线程_  49
    生产者线程_  62
    ......消费者线程_  62
    生产者线程_  94
    ......消费者线程_  94
    生产者线程_  64
    ......消费者线程_  64
    生产者线程_  33
    ......消费者线程_  33
    生产者线程_  41
    ......消费者线程_  41
    生产者线程_  7
    ......消费者线程_  7
    生产者线程_  21
    ......消费者线程_  21

多条生产者和消费者线程：

注意事项：

多条线程需要注意是：

*  我们唤醒的时候需要使用notifyAll()，如果使用notify（）随机唤醒的可能是同一类线程，这样会导致死锁；
 *  需要将if改为while，比如生产者线程有多个，当本生产者线程wait之后，假如另一个生产者线程得到锁（本该消费者得到），如果是if，那么此线程就会继续执行，会导致数据错乱。如果是while则会继续等待。

代码实现：

package ThreadDemo.ThreadExercise;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    
    /*
     * 生产者线程
     */
    class ProducerDemo implements Runnable{
        private List list;
        public ProducerDemo(List list){
            this.list=list;
        }
        @Override
        public void run() {
            while(true){
                Random random=new Random();
                synchronized(list){
                    while(list.size()>0){ 
                        //因为生产者线程有多个，当本线程wait之后，假如一个生产者线程得到锁（本该消费者得到），
                        // 如果是if，那么此线程就会继续执行，会导致数据错乱。
                        //如果是while则会继续等待。
                        try {
                            list.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    list.add(random.nextInt(100));//0-99的随机数；
                    System.out.println(Thread.currentThread().getName()+"  "+list.get(0));
                    list.notifyAll();  //唤醒此对象锁所有等待线程（消费者和生产者线程均有）
                }
            }
        }
    }
    /*
     *  消费者线程
     */
    class ConsumerDemo implements Runnable{
        private List list;
        public ConsumerDemo(List list){
             this.list=list;
        }
        @Override
        public void run() {
            while(true){
                synchronized (list){
                    while(list.size()<1){
                        try {
                            list.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println(Thread.currentThread().getName()+"  "+list.remove(0));
                    list.notifyAll(); //唤醒此对象锁所有等待线程（消费者和生产者线程均有）
                }
            }
        }
    }
    public class ProducerAndConsumer {
        public static void main(String[] args) {
            List list=new ArrayList();
            for (int i = 1; i <4 ; i++) {
                Thread thread1=new Thread(new ProducerDemo(list));
                thread1.setName("生产者线程_"+i+"_");
                Thread thread2=new Thread(new ConsumerDemo(list));
                thread2.setName("......消费者线程_"+i+"_");
                thread2.start();
                thread1.start();
            }
        }
    }

测试结果：

生产者线程_3_  83
    ......消费者线程_1_  83
    生产者线程_1_  74
    ......消费者线程_2_  74
    生产者线程_3_  18
    ......消费者线程_3_  18
    生产者线程_1_  81
    ......消费者线程_2_  81
    生产者线程_3_  56
    ......消费者线程_1_  56
    生产者线程_1_  3
    ......消费者线程_2_  3
    生产者线程_3_  0
    ......消费者线程_3_  0
    生产者线程_1_  18
    ......消费者线程_2_  18
    生产者线程_3_  18
    ......消费者线程_1_  18
    生产者线程_1_  35
    ......消费者线程_2_  35
    生产者线程_3_  81
    ......消费者线程_3_  81
    生产者线程_1_  15
    ......消费者线程_2_  15
    生产者线程_3_  95
    ......消费者线程_1_  95
    生产者线程_1_  2
    ......消费者线程_2_  2

## ReentrantLock+BlockingQueue实现： ##

采用两把锁，实现生产者和消费者同时作业。需要注意的是，生产者的一个锁对象，消费者一个锁对象。分别用来唤醒消费者和生产者

代码实现1：

多条生产者和消费者随机交替，也就是说只要队列没有满那一直生产，只要队列没有空那就一直消费。

package ThreadDemo.ThreadExercise;
    
    import java.util.Queue;
    import java.util.Random;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    //生产者
    class ProducerDemo1 implements Runnable{
        ReentrantLock put;
        ReentrantLock out;
        Condition notFull;
        Condition notEmpty;
        Queue<Integer> queue;
        public ProducerDemo1(ReentrantLock put,ReentrantLock out,Condition notFull,Condition notEmpty,Queue queue){
            this.put=put;
            this.out=out;
            this.notFull=notFull;
            this.notEmpty=notEmpty;
            this.queue=queue;
        }
    
        @Override
        public void run() {
            Random random=new Random();
            while(true){
                put.lock();
                while(queue.size()==10){
                    try {
                        notFull.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //延迟打印速度
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Integer integer=random.nextInt(100);
                System.out.println(Thread.currentThread().getName()+"........."+integer);
                queue.add(integer);
                //队列没有满，通知更多生产者来生产
                if(queue.size()<10){
                    notFull.signal();
                }
                put.unlock();
                //只要生产出一个就通知消费者消费，后续不需要通知
                //因为消费者内部有唤醒更多消费者的机制
                if(queue.size()==1){
                    out.lock();
                    notEmpty.signal();
                    out.unlock();
                }
            }
        }
    }
    class ConsumerDemo1 implements Runnable{
        ReentrantLock put;
        ReentrantLock out;
        Condition notFull;
        Condition notEmpty;
        Queue<Integer> queue;
        public ConsumerDemo1(ReentrantLock put,ReentrantLock out,Condition notFull,Condition notEmpty,Queue queue){
            this.put=put;
            this.out=out;
            this.notFull=notFull;
            this.notEmpty=notEmpty;
            this.queue=queue;
        }
    
        @Override
        public void run() {
            while(true){
                out.lock();
                while(queue.size()==0){
                    try {
                        notEmpty.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"   "+queue.poll());
                //只要队列里还有元素，就通知更多消费者来消费
                if(queue.size()>0){
                    notEmpty.signal();
                }
                out.unlock();
                //只要队列没有满，就通知生产者进行生产
                if(queue.size()==9){
                    put.lock();
                    notFull.signal();
                    put.unlock();
                }
            }
        }
    }
    public class ProducerAndConsumerDemo {
        public static void main(String[] args) {
            LinkedBlockingQueue<Integer> queue=new LinkedBlockingQueue<>();
            ReentrantLock put=new ReentrantLock();
            Condition notFull=put.newCondition();
            ReentrantLock out=new ReentrantLock();
            Condition notEmpty=out.newCondition();
            for (int i = 0; i <3 ; i++) {
                new Thread(new ProducerDemo1(put,out,notFull,notEmpty,queue),"生产者"+i).start();
                new Thread(new ConsumerDemo1(put,out,notFull,notEmpty,queue),"消费者"+i).start();
            }
        }
    }

测试结果：

生产者0.........91
    生产者0.........99
    消费者0   91
    消费者0   99
    生产者1.........99
    生产者1.........12
    消费者1   99
    生产者2.........79
    消费者1   12
    生产者0.........22
    消费者1   79
    消费者1   22
    生产者1.........23
    消费者1   23
    生产者1.........19
    生产者1.........79
    消费者2   19
    消费者2   79
    生产者2.........39
    消费者1   39
    生产者0.........48
    生产者0.........43
    消费者0   48
    生产者1.........64
    消费者0   43
    生产者2.........65
    消费者0   64
    消费者0   65
    生产者0.........24
    生产者0.........72
    消费者0   24
    生产者1.........79
    消费者0   72
    生产者2.........49
    消费者0   79
    消费者0   49
    生产者0.........41
    消费者0   41
    生产者0.........25
    消费者1   25

代码实现2：

多条生产者和消费者，实现生产满了在消费，消费完了在生产。

package ThreadDemo.ThreadExercise;
    
    import java.util.Queue;
    import java.util.Random;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    //生产者
    class ProducerDemo1 implements Runnable{
        ReentrantLock put;
        ReentrantLock out;
        Condition notFull;
        Condition notEmpty;
        Queue<Integer> queue;
        public ProducerDemo1(ReentrantLock put,ReentrantLock out,Condition notFull,Condition notEmpty,Queue queue){
            this.put=put;
            this.out=out;
            this.notFull=notFull;
            this.notEmpty=notEmpty;
            this.queue=queue;
        }
    
        @Override
        public void run() {
            Random random=new Random();
            while(true){
                put.lock();
                while(queue.size()==10){
                    try {
                        notFull.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //延迟打印速度
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Integer integer=random.nextInt(100);
                System.out.println(Thread.currentThread().getName()+"........."+integer);
                queue.add(integer);
                //队列没有满，通知更多生产者来生产
                if(queue.size()<10){
                    notFull.signal();
                }
                put.unlock();
                // 当队列已满时才通知消费者进行消费
                if(queue.size()==10){
                    out.lock();
                    notEmpty.signal();
                    out.unlock();
                }
            }
        }
    }
    class ConsumerDemo1 implements Runnable{
        ReentrantLock put;
        ReentrantLock out;
        Condition notFull;
        Condition notEmpty;
        Queue<Integer> queue;
        public ConsumerDemo1(ReentrantLock put,ReentrantLock out,Condition notFull,Condition notEmpty,Queue queue){
            this.put=put;
            this.out=out;
            this.notFull=notFull;
            this.notEmpty=notEmpty;
            this.queue=queue;
        }
    
        @Override
        public void run() {
            while(true){
                out.lock();
                while(queue.size()==0){
                    try {
                        notEmpty.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"   "+queue.poll());
                //只要队列里还有元素，就通知更多消费者来消费
                if(queue.size()>0){
                    notEmpty.signal();
                }
                out.unlock();
                //当队列为空时，通知生产者进行生产
                if(queue.size()==0){
                    put.lock();
                    notFull.signal();
                    put.unlock();
                }
            }
        }
    }
    public class ProducerAndConsumerDemo {
        public static void main(String[] args) {
            LinkedBlockingQueue<Integer> queue=new LinkedBlockingQueue<>();
            ReentrantLock put=new ReentrantLock();
            Condition notFull=put.newCondition();
            ReentrantLock out=new ReentrantLock();
            Condition notEmpty=out.newCondition();
            for (int i = 0; i <3 ; i++) {
                new Thread(new ProducerDemo1(put,out,notFull,notEmpty,queue),"生产者"+i).start();
                new Thread(new ConsumerDemo1(put,out,notFull,notEmpty,queue),"消费者"+i).start();
            }
        }
    }

测试结果

生产者0.........19
    生产者0.........28
    生产者0.........76
    生产者1.........56
    生产者1.........69
    生产者1.........44
    生产者1.........77
    生产者1.........51
    生产者1.........56
    生产者1.........99
    消费者0   19
    消费者0   28
    消费者0   76
    消费者1   56
    消费者1   69
    消费者1   44
    消费者1   77
    消费者1   51
    消费者1   56
    消费者1   99
    生产者2.........20
    生产者2.........42
    生产者2.........56
    生产者2.........47
    生产者2.........38
    生产者2.........90
    生产者2.........54
    生产者2.........66
    生产者2.........79
    生产者2.........96
    消费者2   20
    消费者2   42
    消费者2   56
    消费者2   47
    消费者2   38
    消费者2   90
    消费者2   54
    消费者2   66
    消费者2   79
    消费者2   96