java并发编程生产者消费者模式

水深无声 2022-11-20 00:59 148阅读 0赞

## 为什么要使用生产者和消费者模式 ##

由于生产与消费是两块功能，所以很难保证两块处理速度完全一致，为了解决这种生产消费能力不均衡的问题，  所以便有了生产者和消费者模式。

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

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯

## 实现方式一使用队列： ##

使用队列，生产者与消费之间通过队列相连，生产者将生产资料放入队列，消费者从队列里面取资源消费。

生产者消费者的示例代码：
    
        生产者：
    
                import java.util.concurrent.BlockingQueue;
    
        public class Producer implements Runnable {
            BlockingQueue<String> queue;
    
            public Producer(BlockingQueue<String> queue) {
                this.queue = queue;
            }
    
            @Override
            public void run() {
                try {
                    String temp = "A Product, 生产线程："
                            + Thread.currentThread().getName();
                    System.out.println("I have made a product:"
                            + Thread.currentThread().getName());
                    queue.put(temp);//如果队列是满的话，会阻塞当前线程
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
        }
    
    
        消费者：
    
                import java.util.concurrent.BlockingQueue;
    
        public class Consumer implements Runnable {
            BlockingQueue<String> queue;
    
            public Consumer(BlockingQueue<String> queue) {
                this.queue = queue;
            }
    
            @Override
            public void run() {
                try {
                    String temp = queue.take();//如果队列为空，会阻塞当前线程
                    System.out.println(temp);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        测试类：
    
                import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
        public class Test3 {
    
            public static void main(String[] args) {
            BlockingQueue<String> queue = new LinkedBlockingQueue<String>(2);
    // BlockingQueue<String> queue = new LinkedBlockingQueue<String>();
    //不设置的话，LinkedBlockingQueue默认大小为Integer.MAX_VALUE
    
    // BlockingQueue<String> queue = new ArrayBlockingQueue<String>(2);
    
                Consumer consumer = new Consumer(queue);
                Producer producer = new Producer(queue);
                for (int i = 0; i < 5; i++) {
                    new Thread(producer, "Producer" + (i + 1)).start();
    
                    new Thread(consumer, "Consumer" + (i + 1)).start();
                }
            }
        }
    
        打印结果：
    
        I have
        made a
        product:Producer1
        I have
        made a
        product:Producer2
        A Product, 生产线程：Producer1
        A Product, 生产线程：Producer2
        I have
        made a
        product:Producer3
        A Product, 生产线程：Producer3
        I have
        made a
        product:Producer5
        I have
        made a
        product:Producer4
        A Product, 生产线程：Producer5
        A Product, 生产线程：Producer4

## 实现方式二阻塞线程： ##

生产者生产产品阻塞线程，结束后唤醒线程，消费者进行消费。

class Restaurant{
    	private Object obj;
    	public void put(){
    		while (true) {
    			synchronized (this) {
    				while (obj != null) {
    					try {
    						wait();
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    				obj = new Object();
    				System.out.print("Order up!  ");
    		        notifyAll();
    			}
    		}
    	}
    	
    	public void get(){
    		while (true) {
    			synchronized (this) {
    				while (obj == null) {
    					try {
    						wait();
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    				
    				System.out.println("waiter get meal");
    				obj = null;
    				notifyAll();
    			}
    			
    		}
    	}
    	
    }
    
    class Chef extends Thread{
    	private Restaurant restaurant;
    	public Chef(Restaurant restaurant){
    		this.restaurant = restaurant;
    	}
    	@Override
    	public void run() {
    		restaurant.put();
    	}
    }
    
    class Waiter extends Thread{
    	private Restaurant restaurant;
    	public Waiter(Restaurant restaurant){
    		this.restaurant = restaurant;
    	}
    	@Override
    	public void run() {
    		restaurant.get();
    	}
    }
    
    public class Test {
    	public static void main(String[] args) {
    		Restaurant restaurant = new Restaurant();
    		Chef produce = new Chef(restaurant);
    		Waiter consumer = new Waiter(restaurant);
    		produce.start();
    		consumer.start();
    	}
    }

参考：

[https://ifeve.com/producers-and-consumers-mode/][https_ifeve.com_producers-and-consumers-mode]

[https://www.cnblogs.com/renjiaqi/p/11324119.html][https_www.cnblogs.com_renjiaqi_p_11324119.html]

[http://www.bubuko.com/infodetail-1058947.html][http_www.bubuko.com_infodetail-1058947.html]

[https_ifeve.com_producers-and-consumers-mode]: https://ifeve.com/producers-and-consumers-mode/
[https_www.cnblogs.com_renjiaqi_p_11324119.html]: https://www.cnblogs.com/renjiaqi/p/11324119.html
[http_www.bubuko.com_infodetail-1058947.html]: http://www.bubuko.com/infodetail-1058947.html