手撕生产者消费者模型

生产者和消费者问题是线程模型中老生常谈的问题，也是面试中经常遇到的问题。今天我们就盘它！！！

一、生产者消费者模型

何为生产者消费者模型呢？

生产者消费者模型是由两类线程构成：

• 生产者线程：“生产”产品，并把产品放到一个缓冲区里；
• 消费者线程：“消费”产品。

如图所示：

小贴士：

• 生产者持续生产，直到缓冲区满，满时阻塞；缓冲区不满后，继续生产；
• 如果缓冲区里的产品从无到有，生产者就得通知一下消费者，告诉它可以来消费了；
• 消费者持续消费，直到缓冲区空，空时阻塞；缓冲区不空后，继续消费；
• 如果缓冲区里的产品从满到不满，消费者也得去通知下生产者，说你可以来生产了。

那我们自己如何实现这个生产者消费者模型呢？

方式一：使用 wait() / notify() 方式

方式二：使用 BlockingQueue 阻塞队列方式

二、wait()/notify() 方式

wait() 和 notify() 都是 Java 中的 Object 类自带的方法，可以用来实现线程间的通信

wait() 方法是用来让当前线程等待，直到有别的线程调用 notify() 将它唤醒，或者我们可以设定一个时间让它自动苏醒。

notify() 方法只能通知一个线程，如果多个线程在等待，那就唤醒任意一个。

notifyAll() 方法是可以唤醒所有等待线程，然后加入同步队列。

具体代码如下：

// 生产者
public class Producer implements Runnable {
	private volatile boolean isRunning = true;
	private final Vector sharedQueue;                            // 内存缓冲区
	private final int SIZE;                                      // 缓冲区大小
	private static AtomicInteger count = new AtomicInteger();    // 总数，原子操作
	private static final int SLEEPTIME = 1000;

	public Producer(Vector sharedQueue, int SIZE) {
		this.sharedQueue = sharedQueue;
		this.SIZE = SIZE;
	}

	@Override
	public void run() {
		int data;
		Random r = new Random();

		System.out.println("start producer id = " + Thread.currentThread().getId());
		try {
			while (isRunning) {
				// 模拟延迟
				Thread.sleep(r.nextInt(SLEEPTIME));

				// 当队列满时阻塞等待
				while (sharedQueue.size() == SIZE) {
					synchronized (sharedQueue) {
						System.out.println("Queue is full, producer " + Thread.currentThread().getId()
								+ " is waiting, size：" + sharedQueue.size());
						sharedQueue.wait();
					}
				}

				// 队列不满时持续创造新元素
				synchronized (sharedQueue) {
					data = count.incrementAndGet();					// 构造任务数据
					sharedQueue.add(data);
					System.out.println("producer create data:" + data + ", size：" + sharedQueue.size());
					sharedQueue.notifyAll();
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			Thread.currentThread().interrupted();
		}
	}

	public void stop() {
		isRunning = false;
	}
}

// 消费者
public class Consumer implements Runnable {

	private final Vector sharedQueue;                            // 内存缓冲区
	private final int SIZE;                                      // 缓冲区大小
	private static final int SLEEPTIME = 1000;

	public Consumer(Vector sharedQueue, int SIZE) {
		this.sharedQueue = sharedQueue;
		this.SIZE = SIZE;
	}

	@Override
	public void run() {

		Random r = new Random();

		System.out.println("start consumer id = " + Thread.currentThread().getId());
		try {
			while (true) {
				// 模拟延迟
				Thread.sleep(r.nextInt(SLEEPTIME));

				// 当队列空时阻塞等待
				while (sharedQueue.isEmpty()) {
					synchronized (sharedQueue) {
						System.out.println("Queue is empty, consumer " + Thread.currentThread().getId()
								+ " is waiting, size：" + sharedQueue.size());
						sharedQueue.wait();
					}
				}

				// 队列不空时持续消费元素
				synchronized (sharedQueue) {
					System.out.println("consumer consume data：" + sharedQueue.remove(0) + ", size：" + sharedQueue.size());
					sharedQueue.notifyAll();
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			Thread.currentThread().interrupt();
		}
	}
}


// 主线程代码
public static void main(String args[]) throws InterruptedException {
	// 1.构建内存缓冲区
	Vector sharedQueue = new Vector();
	int size = 4;

	// 2.建立线程池和线程
	ExecutorService service = Executors.newCachedThreadPool();
	Producer prodThread1 = new Producer(sharedQueue, size);
	Producer prodThread2 = new Producer(sharedQueue, size);
	Producer prodThread3 = new Producer(sharedQueue, size);
	Consumer consThread1 = new Consumer(sharedQueue, size);
	Consumer consThread2 = new Consumer(sharedQueue, size);
	Consumer consThread3 = new Consumer(sharedQueue, size);
	service.execute(prodThread1);
	service.execute(prodThread2);
	service.execute(prodThread3);
	service.execute(consThread1);
	service.execute(consThread2);
	service.execute(consThread3);

	// 3.睡一会儿然后尝试停止生产者
	Thread.sleep(10 * 1000);
	prodThread1.stop();
	prodThread2.stop();
	prodThread3.stop();

	// 4.再睡一会儿关闭线程池
	Thread.sleep(3000);
	service.shutdown();
}

// 执行结果
start producer id = 13
start consumer id = 16
start consumer id = 17
start producer id = 12
start consumer id = 15
start producer id = 14
producer create data:1, size：1
producer create data:2, size：2
consumer consume data：1, size：1

我们解释一下上面的代码吧

• 创造数据：生产者-消费者解决的问题就是数据在多线程间的共享，所以我们首要关心的问题就应该是数据，我们这里采用的是使用一个AtomicInteger类来为我们创造数据，使用它的好处是该类是一个保证原子操作的类，我们使用其中的incrementAndGet()方法不仅能够保证线程安全，还可以达到一个计数的效果，所以是一个既简单又实用的选择，当然也可以使用其他的数据来代替，这里注意的是要保证该类在内存中只存在一份，使用static修饰；
• 内存缓冲区：要保证在多线程环境下内存缓冲区的安全，所以我们考虑使用简单的Vector类来作为我们的内存缓冲区，并且使用final修饰保证内存缓冲区的唯一，然后的话我们需要判断队列是否满，需要手动添加一个标识缓冲区大小的变量SIZE，注意也是final修饰；
• 模拟延迟：这里主要模拟的是一个网络延迟，我们首先定义了一个SLEEPTIME的延迟范围，注意使用的是static final修饰，然后使用Random()类的nextInt()方法来随机选取一个该范围内的值来模拟网络环境中的延迟；
• 停止方法：首先需要知道在Thread类中有一个弃用的stop()方法，我们自己增加一个标志位isRunning来完成我们自己的stop()功能，需要注意的是使用volatile来修饰，保证该标志位的可见性；
• 错误处理：当捕获到错误时，我们应该使用Thread类中的interrupted()方法来终止当前的进程；
• 消息提示：我们主要是要在控制台输出该生产者的信息，包括当前队列的状态，大小，当前线程的生产者信息等，注意的是信息格式的统一；

三、 BlockingQueue 阻塞队列方式

大家还是直接看代码吧：

public class Producer implements Runnable {
	private volatile boolean isRunning = true;
	private BlockingQueue<Integer> queue;                        // 内存缓冲区
	private static AtomicInteger count = new AtomicInteger();    // 总数，原子操作
	private static final int SLEEPTIME = 1000;

	public Producer(BlockingQueue<Integer> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {
		int data;
		Random r = new Random();

		System.out.println("start producer id = " + Thread.currentThread().getId());
		try {
			while (isRunning) {
				// 模拟延迟
				Thread.sleep(r.nextInt(SLEEPTIME));

				// 往阻塞队列中添加数据
				data = count.incrementAndGet();               	// 构造任务数据
				System.out.println("producer " + Thread.currentThread().getId() + " create data：" + data
						+ ", size：" + queue.size());
				if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
					System.err.println("failed to put data：" + data);
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			Thread.currentThread().interrupted();
		}
	}

	public void stop() {
		isRunning = false;
	}
}


public class Consumer implements Runnable {

	private BlockingQueue<Integer> queue;                            // 内存缓冲区
	private static final int SLEEPTIME = 1000;

	public Consumer(BlockingQueue<Integer> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {

		int data;
		Random r = new Random();

		System.out.println("start consumer id = " + Thread.currentThread().getId());
		try {
			while (true) {
				// 模拟延迟
				Thread.sleep(r.nextInt(SLEEPTIME));

				// 从阻塞队列中获取数据
				if (!queue.isEmpty()) {
					data = queue.take();
					System.out.println("consumer " + Thread.currentThread().getId() + " consume data：" + data
							+ ", size：" + queue.size());
				} else {
					System.out.println("Queue is empty, consumer " + Thread.currentThread().getId()
							+ " is waiting, size：" + queue.size());
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			Thread.currentThread().interrupt();
		}
	}
}


public static void main(String args[]) throws InterruptedException {
	// 1.构建内存缓冲区
	BlockingQueue<Integer> queue = new LinkedBlockingDeque<>();

	// 2.建立线程池和线程
	ExecutorService service = Executors.newCachedThreadPool();
	Producer prodThread1 = new Producer(queue);
	Producer prodThread2 = new Producer(queue);
	Producer prodThread3 = new Producer(queue);
	Consumer consThread1 = new Consumer(queue);
	Consumer consThread2 = new Consumer(queue);
	Consumer consThread3 = new Consumer(queue);
	service.execute(prodThread1);
	service.execute(prodThread2);
	service.execute(prodThread3);
	service.execute(consThread1);
	service.execute(consThread2);
	service.execute(consThread3);

	// 3.睡一会儿然后尝试停止生产者
	Thread.sleep(10 * 1000);
	prodThread1.stop();
	prodThread2.stop();
	prodThread3.stop();

	// 4.再睡一会儿关闭线程池
	Thread.sleep(3000);
	service.shutdown();
}

// 运行结果
start producer id = 13
start consumer id = 15
start consumer id = 16
start producer id = 14
start producer id = 12
start consumer id = 17
Queue is empty, consumer 16 is waiting, size：0
Queue is empty, consumer 16 is waiting, size：0
Queue is empty, consumer 16 is waiting, size：0
Queue is empty, consumer 16 is waiting, size：0
producer 13 create data：1, size：0
consumer 15 consume data：1, size：0
producer 12 create data：2, size：0

好了，手撕阻塞队列就讲到这里了！！！我们下期见

引用：

https://www.cnblogs.com/wmyskxz/p/9538177.html#_label0

https://segmentfault.com/a/1190000024444906

https://segmentfault.com/a/1190000018169793?utm_source=sf-similar-article