生产者-消费者模式（阻塞队列实现）

系统管理员 2022-09-24 13:22 203阅读 0赞

生产者消费者模式是并发、多线程编程中经典的设计模式，生产者和消费者通过分离的执行工作解耦，简化了开发模式，生产者和消费者可以以不同的速度生产和消费数据。这篇文章我们来看看什么是生产者消费者模式，这个问题也是多线程面试题中经常被提及的。如何使用阻塞队列（Blocking Queue）解决生产者消费者模式，以及使用生产者消费者模式的好处。

**真实世界中的生产者消费者模式**

生产者和消费者模式在生活当中随处可见，它描述的是协调与协作的关系。比如一个人正在准备食物（生产者），而另一个人正在吃（消费者），他们使用一个共用的桌子用于放置食物和取走食物，生产者准备食物，如果桌子上已经满了就等待，如果桌子空了的话消费者（那个吃的）等待。这里桌子就是一个共享的对象。在Java Executor框架自身实现了生产者消费者模式它们分别负责添加和执行任务。

**生产者消费者模式的好处**

它的确是一种实用的设计模式，常用于编写多线程或并发代码。下面是它的一些优点：

1.   它简化的开发，你可以独立地或并发的编写消费者和生产者，它仅仅只需知道共享对象是谁
2.   生产者不需要知道谁是消费者或者有多少消费者，对消费者来说也是一样
3.   生产者和消费者可以以不同的速度执行
4.   分离的消费者和生产者在功能上能写出更简洁、可读、易维护的代码

**多线程中的生产者消费者问题**

生产者消费者问题是一个流行的面试题，面试官会要求你实现生产者消费者设计模式，以至于能让生产者应等待如果队列或篮子满了的话，消费者等待如果队列或者篮子是空的。这个问题可以用不同的方式来现实，经典的方法是使用wait和notify方法在生产者和消费者线程中合作，在队列满了或者队列是空的条件下阻塞，Java5的阻塞队列（BlockingQueue）数据结构更简单，因为它隐含的提供了这些控制，现在你不需要使用wait和nofity在生产者和消费者之间通信了，阻塞队列的put()方法将阻塞如果队列满了，队列take()方法将阻塞如果队列是空的。在下部分我们可以看到代码例子。

**使用阻塞队列实现生产者消费者模式**

阻塞队列实现生产者消费者模式超级简单，它提供开箱即用支持阻塞的方法put()和take()，开发者不需要写困惑的wait、nofity代码去实现通信。BlockingQueue 一个接口，Java5提供了不同的现实，如ArrayBlockingQueue和LinkedBlockingQueue，两者都是先进先出（FIFO）顺序。而ArrayLinkedQueue是自然有界的，LinkedBlockingQueue可选的边界。下面这是一个完整的生产者消费者代码例子，对比传统的wait、nofity代码，它更易于理解。

put()方法：类似于我们上面的生产者线程，容量达到最大时，自动阻塞。

take()方法：类似于我们上面的消费者线程，容量为0时，自动阻塞。

/**
     * 仓库类Storage实现缓冲区
     * 
     */
    public class Storage {
    
    	// 仓库存储的载体 （容量为6）
    	private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(6);
    
    	// 生产num个产品
    	public void produce() {
    
    		int number = 0;
    		
    		while (number < 10) {
    			try {
    				// 放入产品，容量满的时候自动阻塞
    				list.put(number);
    				System.out.println("生产1个\t【库存量】" + list.size());
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    
    			number++;
    		}
    	}
    
    	// 消费num个产品
    	public void consume() {
    
    		while (true) {
    			try {
    				// 消费产品，容量为空的时候自动阻塞
    				System.out.println("消费1个:" + list.take() + "\t【库存量】"
    						+ list.size());
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

public class Producer extends Thread {
    
    	// 所在放置的仓库
    	private Storage storage;
    
    	// 构造函数，设置仓库
    	public Producer(Storage storage) {
    		this.storage = storage;
    	}
    
    	@Override
    	public void run() {
    		// TODO Auto-generated method stub
    		storage.produce();
    	}
    }

public class Consumer extends Thread {
    
    	// 所在放置的仓库
    	private Storage storage;
    
    	// 构造函数，设置仓库
    	public Consumer(Storage storage) {
    		this.storage = storage;
    	}
    
    	@Override
    	public void run() {
    		// TODO Auto-generated method stub
    		storage.consume();
    	}
    }

public class Test {
    
    	public static void main(String[] args) {
    
    		// 仓库对象
    		Storage storage = new Storage();
    
    		// 生产者对象
    		Producer p1 = new Producer(storage);
    
    		// 消费者对象
    		Consumer c1 = new Consumer(storage);
    
    		// 线程开始执行
    		c1.start();
    		p1.start();
    	}
    }

OutPut：

生产1个	【库存量】1
    消费1个:0	【库存量】0
    生产1个	【库存量】1
    生产1个	【库存量】1
    消费1个:1	【库存量】0
    生产1个	【库存量】2
    消费1个:2	【库存量】1
    生产1个	【库存量】2
    消费1个:3	【库存量】1
    生产1个	【库存量】2
    消费1个:4	【库存量】1
    消费1个:5	【库存量】1
    消费1个:6	【库存量】0
    生产1个	【库存量】2
    生产1个	【库存量】1
    生产1个	【库存量】2
    生产1个	【库存量】3
    消费1个:7	【库存量】2
    消费1个:8	【库存量】1
    消费1个:9	【库存量】0

有时使用BlockingQueue可能会出现put()和System.out.println()输出不匹配的情况，这是由于它们之间没有同步造成的。当缓冲区已满，生产者在put()操作时，put()内部调用了await()方法，放弃了线程的执行，然后消费者线程执行，调用take()方法，take()内部调用了signal()方法，通知生产者线程可以执行，致使在消费者的println()还没运行的情况下生产者的println()先被执行，所以有了输出不匹配的情况