实现生产者-消费者模式的五种方式

MOKE 2019-08-23 PM 10℃ 0条

在学习完 RabbitMQ 后,才明白为什么那么喜欢问生产者-消费者问题,因为在很多开发领域都能派上用场。
Java基础 以及 操作系统 都有接触,但还是要在这里再总结一下。

实现方式

生产者消费者问题是线程模型中的经典问题:生产者和消费者在同一时间段内共用同一存储空间,生产者向空间里生产数据,而消费者取走数据。

保证同一时刻只有一个线程读或写:

  • 用synchronized对存储加锁,然后用object原生的wait() 和 notify()做同步
  • 用concurrent.locks.Lock,然后用condition的await() 和signal()做同步
  • 使用信号量semaphore

保证存储内部的读写唯一

  • 直接使用concurrent.BlockingQueue
  • 使用PipedInputStream/PipedOutputStream

synchronized、wait()、notify()

  • wait():当缓冲区已满/空时,生产者/消费者线程停止自己的执行,放弃锁,使自己处于等待状态,让其他线程执行。
  • notify():当生产者/消费者向缓冲区放入/取出一个产品时,向其他等待的线程发出可执行的通知,同时放弃锁,使自己处于等待状态。

实现

        public class Main {
            static int i = 0;
            static int size = 10;
            Queue<String> queue = new LinkedList<String>();
            
            public static void main(String[] args) {
                Main main = new Main();
                main.new Producter("producter[1]").start();
                main.new Producter("producter[2]").start();
                main.new Consumer("consumer[1]").start();
                main.new Consumer("consumer[2]").start();
                main.new Consumer("consumer[3]").start();
        
            }
            
            //生产者
            class Producter extends Thread{
                private String name;
                
                public Producter(String name) {
                    this.name = name;
                }
        
                @Override
                public void run() {
                    while(true){
                        synchronized (queue) {
                            while(queue.size()==size){
                                System.out.println("The queue is full! "+name+" is waiting for Consumer");
                                try {
                                    queue.wait();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }
                            System.out.println(name+":product====>production_"+i);
                            queue.offer("production_" + i);
                            i++;
                            queue.notifyAll();
                        }
                        try {
                            sleep(new Random().nextInt(1000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            
            //消费者
            class Consumer extends Thread{
                private String name;
                    
                public Consumer(String name) {
                    this.name = name;
                }
                    
                @Override
                public void run() {
                    while(true){
                        synchronized (queue) {
                            while(queue.isEmpty()){
                                System.out.println("The queue is empty! "+name+" is waiting for Producter");
                                try {
                                    queue.wait();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }
                            String s = queue.poll();
                            System.out.println(name+":consume====>"+s);
                            queue.notifyAll();
                        }
                        try {
                            sleep(new Random().nextInt(3500));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }

结果
在这里插入图片描述


Lock、Condition、await()、signal()

实现

        public class Main {
            static int i = 0;
            static int size = 10;
            Queue<String> queue = new LinkedList<String>();
            Lock lock = new ReentrantLock();
            Condition condition = lock.newCondition();
            
            public static void main(String[] args) {
                Main main = new Main();
                main.new Producter("producter[1]").start();
                main.new Producter("producter[2]").start();
                main.new Consumer("consumer[1]").start();
                main.new Consumer("consumer[2]").start();
                main.new Consumer("consumer[3]").start();
        
            }
            
            //生产者
            class Producter extends Thread{
                private String name;
                
                public Producter(String name) {
                    this.name = name;
                }
        
                @Override
                public void run() {
                    while(true){
                        lock.lock();
                        while(queue.size()==size){
                            System.out.println("The queue is full! "+name+" is waiting for Consumer");
                            try {
                                condition.await();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        System.out.println(name+":product====>production_"+i);
                        queue.offer("production_" + i);
                        i++;
                        condition.signalAll();;
                        lock.unlock();
                        try {
                            sleep(new Random().nextInt(1000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
    
    //消费者
    class Consumer extends Thread{
        private String name;
            
        public Consumer(String name) {
            this.name = name;
        }
            
        @Override
        public void run() {
            while(true){
                lock.lock();
                while(queue.isEmpty()){
                    System.out.println("The queue is empty! "+name+" is waiting for Producter");
                    try {
                        condition.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                String s = queue.poll();
                System.out.println(name+":consume====>"+s);
                condition.signalAll();
                lock.unlock();
                try {
                    sleep(new Random().nextInt(4000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

结果
在这里插入图片描述


BlockingQueue

并发编程 中,有提到在 J.U.C 中有一个接口 BlockingQueue,有一下几个实现的阻塞队列:

  • java.util.concurrent.ArrayBlockingQueue
  • java.util.concurrent.LinkedBlockingQueue
  • java.util.concurrent.SynchronousQueue
  • java.util.concurrent.PriorityBlockingQueue

我们可以使用 FIFO队列,如 ArrayBlockingQueue 或者 LinkedBlockingQueue 来实现生产线-消费者模型,这里我们使用 LinkedBlockingQueue。

LinkedBlockingQueue
可以从源码看到,他是在队列的基础上,使用 Lock 实现了同步队列,我们主要使用两个方法:

  • put()方法:容量达到最大时,自动阻塞。
  • take()方法:容量为0时,自动阻塞
            /** Main lock guarding all access */
            final ReentrantLock lock = new ReentrantLock();
             
            /** Condition for waiting takes */
            private final Condition notEmpty = lock.newCondition();
             
            /** Condition for waiting puts */
            private final Condition notFull = lock.newCondition();
             
            public void put(E e) throws InterruptedException {
                putLast(e);
            }
             
            public void putLast(E e) throws InterruptedException {
                if (e == null) throw new NullPointerException();
                Node<E> node = new Node<E>(e);
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                    while (!linkLast(node))
                        notFull.await();
                } finally {
                    lock.unlock();
                }
            }

实现

        public class Main {
            static int i = 0;
            static int size = 10;
            BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>(size);//指定阻塞队列长度
            
            public static void main(String[] args) {
                Main main = new Main();
                main.new Producter("producter[1]").start();
                main.new Producter("producter[2]").start();
                main.new Consumer("consumer[1]").start();
                main.new Consumer("consumer[2]").start();
                main.new Consumer("consumer[3]").start();
        
            }
            
            //生产者
            class Producter extends Thread{
                private String name;
                
                public Producter(String name) {
                    this.name = name;
                }
        
                @Override
                public void run() {
                    while(true){
                        try {
                            //这个是为了不产生相同的 "production_"+i,也可以不加,不过会产生名称相同的产品
                            synchronized (blockingQueue) {
                                blockingQueue.put("production_"+i);
                                System.out.println(name+" product====>production_"+i);
                                i++;
                            }
                            sleep(new Random().nextInt(1000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            
            //消费者
            class Consumer extends Thread{
                private String name;
                    
                public Consumer(String name) {
                    this.name = name;
                }
                    
                @Override
                public void run() {
                    while(true){
                        try {
                            System.out.println(name+" consume====>"+blockingQueue.take());
                            sleep(new Random().nextInt(1000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }

结果
在这里插入图片描述


Semaphore

Semaphore 信号量,可以控制对互斥资源的访问的线程数,使用在 并发编程 中有,这里不再说明。

实现

        public class Main {
            int i = 0;
            Queue<String> queue = new LinkedList<String>();
            
            Semaphore notFull = new Semaphore(10);//可以生产的数量,生产一个少一个 permit
            Semaphore notEmpty = new Semaphore(0);//可以消费的数量,消费一个多一个 permit
            Semaphore mutex = new Semaphore(1);//控制 queue 的互斥访问
            
            //生产者
            class Producter extends Thread{
                private String name;
                
                public Producter(String name) {
                    this.name = name;
                }
        
                @Override
                public void run() {
                    while(true){
                        try {
                            notFull.acquire();
                            mutex.acquire();
                            queue.offer("production_"+i);
                            System.out.println(name+":product====>production_"+i);
                            i++;
                            mutex.release();
                            notEmpty.release();
                            sleep(new Random().nextInt(1000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            
            //消费者
            class Consumer extends Thread{
                private String name;
                    
                public Consumer(String name) {
                    this.name = name;
                }
                    
                @Override
                public void run() {
                    while(true){
                        try {
                            notEmpty.acquire();
                            mutex.acquire();
                            String s = queue.poll();
                            System.out.println(name+":consume====>"+s);
                            mutex.release();
                            notFull.release();
                            sleep(new Random().nextInt(1000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }

            public static void main(String[] args) {
                Main main = new Main();
                main.new Producter("producter[1]").start();
                main.new Producter("producter[2]").start();
                main.new Consumer("consumer[1]").start();
                main.new Consumer("consumer[2]").start();
                main.new Consumer("consumer[3]").start();
            }
        }

结果
在这里插入图片描述


PipedInputStream、PipedOutputStream

在这里插入图片描述在这里插入图片描述

实现

        public class Main {
            static int i = 0;
            static int size = 10;
            PipedInputStream pis = new PipedInputStream(size);
            PipedOutputStream pos = new PipedOutputStream();
            {
                try {
                    pis.connect(pos);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            
            public static void main(String[] args) {
                Main main = new Main();
                main.new Producter("producter[1]").start();
                main.new Producter("producter[2]").start();
                main.new Consumer("consumer[1]").start();
                main.new Consumer("consumer[2]").start();
                main.new Consumer("consumer[3]").start();
        
            }
            
            //生产者
            class Producter extends Thread{
                private String name;
                
                public Producter(String name) {
                    this.name = name;
                }
        
                @Override
                public void run() {
                    while(true){
                        try {
                            synchronized (pos) {//避免产生相同名称的产品,如果不在意可以去掉
                                pos.write(i);
                                pos.flush();
                                System.out.println(name+":product====>production_"+i);
                                i++;
                            }
                            sleep(new Random().nextInt(1000));
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            
            //消费者
            class Consumer extends Thread{
                private String name;
                    
                public Consumer(String name) {
                    this.name = name;
                }
                    
                @Override
                public void run() {
                    while(true){
                        try {
                            System.out.println(name+":consume====>production_"+pis.read());
                            sleep(new Random().nextInt(3500));
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }

结果
在这里插入图片描述

标签: Java

非特殊说明,本博所有文章均为博主原创。

评论啦~