Java多线程（2）生产者消费者问题（一）

一、问题描述

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。如何用代码描述此类问题。

二、一个消费者线程、一个生产者线程

有几点需要强调：

1.main方法中的资源res的线程t1、t2共享的。所以当操作该资源时需要同步

2.同步函数的锁即为this，所以res即为线程t1、t2的锁

3.因为只有两个线程所以能够准确的唤醒对象线程（即消费者线程唤醒生产者进程、生产者进程唤醒消费者进程），

那么问题就来了，如果是多消费者进程，多生产进程怎么办？

package com.thread.pcprob;

/**
* 共享数据
* @author dqf
*
*/
public class Resource {
//产品编号
private Integer proNo = 0;
//判断是消费数据，还是生产数据
private boolean flag = false;

//生产数据
public synchronized void produce(){
if(flag){//如果flag=true,则等待消费；
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("生产-->" + (++proNo));
flag = true;
this.notify();//唤醒等待进程
}

//消费数据
public synchronized void consume(){
if(!flag){//如果flag=false,则等待生产
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("消费-------->" + proNo);
flag = false;
this.notify();//唤醒等待进程
}
}

package com.thread.pcprob;

/**
* 生产者
* @author dqf
*
*/
public class Producer implements Runnable{

private Resource res;

public Producer(Resource res) {
this.res = res;
}

@Override
public void run() {
while(true){
res.produce();
}
}
}

package com.thread.pcprob;

/**
* 消费者
* @author dqf
*
*/
public class Consumer implements Runnable{

private Resource res;

public Consumer(Resource res) {
this.res = res;
}

@Override
public void run() {
while(true){
res.consume();
}
}
}

package com.thread.pcprob;

public class App {
public static void main(String[] args) {
Resource res = new Resource();

Producer p = new Producer(res);
Consumer c = new Consumer(res);

Thread t1 = new Thread(p);
Thread t2 = new Thread(c);

t1.start();
t2.start();

}
}

部分输出：

消费-------->159091
生产-->159092
消费-------->159092
生产-->159093
消费-------->159093
生产-->159094
消费-------->159094
生产-->159095
消费-------->159095
生产-->159096
消费-------->159096
生产-->159097
消费-------->159097
生产-->159098
消费-------->159098
生产-->159099
消费-------->159099

三、多生产者进程、多消费者进程

我们把main方法修改如下：

public class App {
public static void main(String[] args) {
Resource res = new Resource();

Producer p = new Producer(res);
Consumer c = new Consumer(res);

Thread t1 = new Thread(p);
Thread t2 = new Thread(p);

Thread t3 = new Thread(c);
Thread t4 = new Thread(c);

t1.start();
t2.start();
t3.start();
t4.start();

}
}

部分输出：

消费-------->123384
消费-------->123384
生产-->123385
消费-------->123385
消费-------->123385
生产-->123386
消费-------->123386
消费-------->123386
生产-->123387
消费-------->123387
消费-------->123387
生产-->123388
消费-------->123388

我们发现会出现两种错误的情况

1.生产一个，消费多次

2.生产多次，消费一次

下面来分析产生两种问题的原因：



假设一种状态：

flag = false，t1获取到锁，t2、t3、t4等待锁释放。

因为flag为false，t1阻塞自动释放锁，t2获取锁，同样因为flag为false，t2阻塞自动释放锁。

若t3获取锁，执行到this.notify()，则会唤醒一个阻塞状态的线程，如果唤醒了t1线程，t1线程结束后又唤醒了t2。

则，就会出现多次生产，一次消费的情况。

其实造成这种情况的主要原因是：不能指定具体唤醒哪一个线程、或者说不能指定唤醒哪一类线程。导致，生产进程结束后，又唤醒了生产进程；或者消费进程结束后，又唤醒消费进程。

如果将Resource改为（if改为while），那么在线程被唤醒后，都会再次判断一次flag。

package com.thread.pcprob;

/**
* 共享数据
* @author dqf
*
*/
public class Resource {
//产品编号
private Integer proNo = 0;
//判断是消费数据，还是生产数据
private boolean flag = false;

//生产数据
public synchronized void produce(){
while(flag){//如果flag=true,则等待消费；
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("生产-->" + (++proNo));
flag = true;
this.notify();//唤醒等待进程
}

//消费数据
public synchronized void consume(){
while(!flag){//如果flag=false,则等待生产
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("消费-------->" + proNo);
flag = false;
this.notify();//唤醒等待进程
}
}

但是这种情况又会导致，t1、t2、t3、t4全部处于阻塞状态。

最终写法（notify改为notifyAll）：

package com.thread.pcprob;

/**
* 共享数据
* @author dqf
*
*/
public class Resource {
//产品编号
private Integer proNo = 0;
//判断是消费数据，还是生产数据
private boolean flag = false;

//生产数据
public synchronized void produce(){
while(flag){//如果flag=true,则等待消费；
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("生产-->" + (++proNo));
flag = true;
this.notifyAll();//唤醒等待进程
}

//消费数据
public synchronized void consume(){
while(!flag){//如果flag=false,则等待生产
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("消费-------->" + proNo);
flag = false;
this.notifyAll();//唤醒等待进程
}
}