手撕生产者消费者模型


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

一、生产者消费者模型

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

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

  • 生产者线程:“生产”产品,并把产品放到一个缓冲区里;
  • 消费者线程:“消费”产品。

如图所示:

img

小贴士:

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

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

方式一:使用 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


文章作者: Gtwff
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 Gtwff !
  目录