- 一、生产者消费者模式
- 二、代码实现
- （一）synchronized + wait notifyAll 方式实现
- 1、代码
- 2、执行情况
- （二）Lock Condition + signalAll await方式实现
- 1、代码
- 2、执行情况
- （三）LinkedBlockingDeque 方式实现
- 1、代码
- 2、执行情况
题目：
有一个仓库，容量为N；有若干个生产者，将生产的产品存入到仓库中，除非仓库已满；又有若干消费者，从仓库中取出产品消费，除非仓库已空。请写一段程序，让这个模式正常运转，考虑到多线程并发情况。
public class ProducerConsumerRun {
public static void main(String[] args) {
Resource resource = new Resource();
ProducerThread p1 = new ProducerThread(resource);
ProducerThread p2 = new ProducerThread(resource);
ProducerThread p3 = new ProducerThread(resource);
ConsumerThread c1 = new ConsumerThread(resource);
p1.start();
p2.start();
p3.start();
c1.start();
}
}
public class Resource {
private int num = 0; //资源计数
private int size = 10; //资源池上限
public synchronized void remove(){
if(num>0){
num--;
System.out.println("消费者线程:"+Thread.currentThread().getName()+"消耗一个资源，当前资源剩余:"+num);
notifyAll(); //唤醒所有等待线程
}else{
try {
wait();
System.out.println("消费者线程:"+Thread.currentThread().getName()+"当前无可消费资源，进入等待状态");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public synchronized void add(){
if(num
num++;
System.out.println("生产者线程:"+Thread.currentThread().getName()+"生产一个资源，当前资源剩余:"+num);
notifyAll(); //唤醒所有等待线程
}else{
try {
wait();
System.out.println("生产者线程:"+Thread.currentThread().getName()+"当前不能生产资源，资源池已满，线程进入等待状态");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class ProducerThread extends Thread {
private Resource resource;
public ProducerThread(Resource resource){
this.resource = resource;
}
@Override
public void run(){
while(true){
try {
Thread.sleep(1000); //一秒钟生产一个资源
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.add();
}
}
}
public class ConsumerThread extends Thread {
private Resource resource;
public ConsumerThread(Resource resource){
this.resource = resource;
}
@Override
public void run(){
while(true){
try {
Thread.sleep(1000); //一秒钟消耗一个资源
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.remove();
}
}
}
2、执行情况
从打印情况来看，当资源池满后，生产者线程进入等待时，消费者消费一个资源后，所有处于等待中的生产者都被唤醒了，将程序中的notifyAll 换成notify，会唤醒等待中的一个线程。
生产者线程:Thread-0生产一个资源，当前资源剩余:1 消费者线程:Thread-3消耗一个资源，当前资源剩余:0 生产者线程:Thread-1生产一个资源，当前资源剩余:1 生产者线程:Thread-2生产一个资源，当前资源剩余:2 生产者线程:Thread-0生产一个资源，当前资源剩余:3 消费者线程:Thread-3消耗一个资源，当前资源剩余:2 生产者线程:Thread-1生产一个资源，当前资源剩余:3 生产者线程:Thread-2生产一个资源，当前资源剩余:4 生产者线程:Thread-0生产一个资源，当前资源剩余:5 消费者线程:Thread-3消耗一个资源，当前资源剩余:4 生产者线程:Thread-1生产一个资源，当前资源剩余:5 生产者线程:Thread-2生产一个资源，当前资源剩余:6 生产者线程:Thread-0生产一个资源，当前资源剩余:7 消费者线程:Thread-3消耗一个资源，当前资源剩余:6 生产者线程:Thread-1生产一个资源，当前资源剩余:7 生产者线程:Thread-2生产一个资源，当前资源剩余:8 生产者线程:Thread-0生产一个资源，当前资源剩余:9 消费者线程:Thread-3消耗一个资源，当前资源剩余:8 生产者线程:Thread-1生产一个资源，当前资源剩余:9 生产者线程:Thread-2生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-0当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-0生产一个资源，当前资源剩余:10 生产者线程:Thread-2当前不能生产资源，资源池已满，线程进入等待状态 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-2当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-0当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-1当前不能生产资源，资源池已满，线程进入等待状态 消费者线程:Thread-3消耗一个资源，当前资源剩余:8 生产者线程:Thread-2生产一个资源，当前资源剩余:9 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-0当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-2生产一个资源，当前资源剩余:10（二）Lock Condition + signalAll await方式实现 1、代码
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ProducerConsumerRun {
public static void main(String[] args) {
Lock lock = new ReentrantLock();
Condition producerCondition = lock.newCondition();
Condition consumerCondition = lock.newCondition();
Resource resource = new Resource(lock,producerCondition,consumerCondition);
ProducerThread p1 = new ProducerThread(resource);
ProducerThread p2 = new ProducerThread(resource);
ProducerThread p3 = new ProducerThread(resource);
ConsumerThread c1 = new ConsumerThread(resource);
p1.start();
p2.start();
p3.start();
c1.start();
}
}
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
public class Resource {
private int num = 0;
private int size = 10;
private Lock lock;
private Condition producerCondition;
private Condition consumerCondition;
public Resource(Lock lock, Condition producerCondition, Condition consumerCondition){
this.lock = lock;
this.producerCondition = producerCondition;
this.consumerCondition = consumerCondition;
}
public void remove(){
lock.lock();
try {
if (num > 0) {
num--;
System.out.println("消费者线程:" + Thread.currentThread().getName() + "消耗一个资源，当前资源剩余:" + num);
producerCondition.signalAll();
} else {
try {
consumerCondition.await();
System.out.println("消费者线程:" + Thread.currentThread().getName() + "当前无可消费资源，进入等待状态");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}finally {
lock.unlock();
}
}
public void add(){
lock.lock();
try {
if (num < size) {
num++;
System.out.println("生产者线程:" + Thread.currentThread().getName() + "生产一个资源，当前资源剩余:" + num);
consumerCondition.signalAll();
} else {
try {
producerCondition.await();
System.out.println("生产者线程:" + Thread.currentThread().getName() + "当前不能生产资源，资源池已满，线程进入等待状态");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}finally {
lock.unlock();
}
}
}
public class ProducerThread extends Thread {
private Resource resource;
public ProducerThread(Resource resource){
this.resource = resource;
}
@Override
public void run(){
while(true){
try {
Thread.sleep((long) (1000*Math.random()));
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.add();
}
}
}
public class ConsumerThread extends Thread {
private Resource resource;
public ConsumerThread(Resource resource){
this.resource = resource;
}
@Override
public void run(){
while(true){
try {
Thread.sleep((long) (1000*Math.random()));
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.remove();
}
}
}
2、执行情况
生产者线程:Thread-2生产一个资源，当前资源剩余:1 生产者线程:Thread-0生产一个资源，当前资源剩余:2 生产者线程:Thread-1生产一个资源，当前资源剩余:3 生产者线程:Thread-2生产一个资源，当前资源剩余:4 消费者线程:Thread-3消耗一个资源，当前资源剩余:3 生产者线程:Thread-2生产一个资源，当前资源剩余:4 生产者线程:Thread-2生产一个资源，当前资源剩余:5 生产者线程:Thread-0生产一个资源，当前资源剩余:6 消费者线程:Thread-3消耗一个资源，当前资源剩余:5 生产者线程:Thread-2生产一个资源，当前资源剩余:6 生产者线程:Thread-0生产一个资源，当前资源剩余:7 生产者线程:Thread-1生产一个资源，当前资源剩余:8 生产者线程:Thread-2生产一个资源，当前资源剩余:9 生产者线程:Thread-2生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-1当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-0当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-2当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-2当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-1当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-0生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-1当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-2当前不能生产资源，资源池已满，线程进入等待状态 生产者线程:Thread-0生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9`（三）LinkedBlockingDeque 方式实现 1、代码
public class ProducerConsumerRun {
public static void main(String[] args) {
Resource resource = new Resource();
ProducerThread p1 = new ProducerThread(resource);
ProducerThread p2 = new ProducerThread(resource);
ProducerThread p3 = new ProducerThread(resource);
ConsumerThread c1 = new ConsumerThread(resource);
p1.start();
p2.start();
p3.start();
c1.start();
}
}
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
public class Resource {
private BlockingQueue queue = new LinkedBlockingDeque(10);
public void remove(){
try {
queue.take();
System.out.println("消费者线程:" + Thread.currentThread().getName() + "消耗一个资源，当前资源剩余:" + queue.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void add(){
try {
queue.put(1);
System.out.println("生产者线程:" + Thread.currentThread().getName() + "生产一个资源，当前资源剩余:" + queue.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public class ProducerThread extends Thread {
private Resource resource;
public ProducerThread(Resource resource){
this.resource = resource;
}
@Override
public void run(){
while(true){
try {
Thread.sleep((long) (1000*Math.random()));
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.add();
}
}
}
public class ConsumerThread extends Thread {
private Resource resource;
public ConsumerThread(Resource resource){
this.resource = resource;
}
@Override
public void run(){
while(true){
try {
Thread.sleep((long) (1000*Math.random()));
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.remove();
}
}
}
2、执行情况
生产者线程:Thread-0生产一个资源，当前资源剩余:1 消费者线程:Thread-3消耗一个资源，当前资源剩余:0 生产者线程:Thread-1生产一个资源，当前资源剩余:1 消费者线程:Thread-3消耗一个资源，当前资源剩余:0 生产者线程:Thread-1生产一个资源，当前资源剩余:1 生产者线程:Thread-2生产一个资源，当前资源剩余:2 生产者线程:Thread-1生产一个资源，当前资源剩余:3 生产者线程:Thread-0生产一个资源，当前资源剩余:4 消费者线程:Thread-3消耗一个资源，当前资源剩余:3 生产者线程:Thread-1生产一个资源，当前资源剩余:4 生产者线程:Thread-2生产一个资源，当前资源剩余:5 生产者线程:Thread-1生产一个资源，当前资源剩余:6 生产者线程:Thread-1生产一个资源，当前资源剩余:7 消费者线程:Thread-3消耗一个资源，当前资源剩余:6 生产者线程:Thread-0生产一个资源，当前资源剩余:7 生产者线程:Thread-1生产一个资源，当前资源剩余:8 生产者线程:Thread-2生产一个资源，当前资源剩余:9 生产者线程:Thread-0生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-2生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-0生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-2生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-1生产一个资源，当前资源剩余:10 消费者线程:Thread-3消耗一个资源，当前资源剩余:9 生产者线程:Thread-0生产一个资源，当前资源剩余:10
参照文档：
生产者-消费者模式的三种实现方式：https://blog.csdn.net/datase/article/details/80070966
