前言
生产者-消费者模式是一个经典的多线程设计模式。
在生产者-消费者模式中，通常有两类线程，即若干个生产者和消费者线程。
- 生产者线程负责提交用户请求
- 消费者线程负责处理生产者提交的任务。
- 内存缓冲区 缓存生产者提交的任务或数据，供消费者使用。
开发需要解决的问题：
- 生产者线程与消费者线程对内存缓冲区的操作的线程安全问题。
- 虚假唤醒。
测试：
/**
* 生产者与消费者案例。
* @author
*/
public class TestProductorAndConsumer {
public static void main(String[] args) {
//创建职员
Clerk clerk = new Clerk();
//创建生产者与消费者线程
Productor productor = new Productor(clerk);
Consumer consumer = new Consumer(clerk);
new Thread(productor, "生产者1").start();
new Thread(consumer, "消费者1").start();
new Thread(productor, "生产者2").start();
new Thread(consumer, "消费者2").start();
}
}
// Clerk职员
class Clerk {
private int product = 0; //产品数量
private int capacity = 4; // 容量
// 进货
public synchronized void get() {
if (product >= capacity) {
System.out.println("产品已满！");
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
System.out.println(Thread.currentThread().getName() + " : " + ++product);
this.notifyAll();
}
}
// 卖货
public synchronized void sale() {
if (product <= 0) {
System.out.println("缺货！");
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
System.out.println(Thread.currentThread().getName() + " : " + --product);
this.notifyAll();
}
}
}
// 生产者
class Productor implements Runnable {
private Clerk clerk;
public Productor(Clerk clerk) {
this.clerk = clerk;
}
@Override
public void run() {
for (int i = 0; i < 20; i++) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
clerk.get();
}
}
}
// 消费者
class Consumer implements Runnable {
private Clerk clerk;
public Consumer(Clerk clerk) {
this.clerk = clerk;
}
@Override
public void run() {
for (int i = 0; i < 20; i++) {
clerk.sale();
}
}
}
这就是一个简单的生产者与消费者模型。
我们在Clerk类中给get()方法和sale()方法加了synchronized修饰符，来保证线程同步。
但是运行后发现程序并没有运行结束，分析发现，我们的生产者线程最后没有被唤醒，导致程序没有结束。
对程序做一下修改：
/**
* 对Clerk类的get()与sale()方法做一点修改
*/
public synchronized void get() {
if (product >= capacity) {
System.out.println("产品已满！");
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + " : " + ++product);
this.notifyAll();
}
// 卖货
public synchronized void sale() {
if (product <= 0) {
System.out.println("缺货！");
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + " : " + --product);
this.notifyAll();
}
把notifyAll() 方法提到了else外面，保证每个线程结束都能调用notifyAll()方法，运行一下，发现程序确实能结束，但是程序 product 变成了负数。这是由于调用notifyAll()唤醒了消费者模型，执行–product导致。
我们来看一下wait()这个方法：
这就是我们要解决的虚假唤醒问题！！！。
文档提醒我们使用循环。再对程序做一点修改
// 进货
public synchronized void get() {
// 使用while防止虚假唤醒
while(product >= capacity) {
System.out.println("产品已满！");
try {
// 在一个参数版本中，中断和虚假的唤醒是可能的，这个方法应该总是在循环中使用:
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 除非product<capacity,否则不执行++product操作
System.out.println(Thread.currentThread().getName() + " : " + ++product);
this.notifyAll();
}
// 卖货
public synchronized void sale() {
while (product <= 0) {
System.out.println("缺货！");
try {
// 在一个参数版本中，中断和虚假的唤醒是可能的，这个方法应该总是在循环中使用:
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 除非product>capacity,否则不执行--product操作
System.out.println(Thread.currentThread().getName() + " : " + --product);
this.notifyAll();
}
将 if 改为 while 循环后，
生产者线程被唤醒后进行判断：如果product >= capacity，则继续调用wait()等待，直到再次被唤醒。如果product < capacity, 则执行++product。
同理消费者线程被唤醒后也会进行判断，不满足条件会继续等待，直到再次被唤醒。满足条件后处理任务。
至此，我们的生产者-消费者模型就圆满完成了。
我们再对程序做一点修改，不使用synchronized来修饰方法，而是采用可重入锁ReentrantLock来手动加锁与释放锁。此时我们也就不能再使用wait()和notifyAll()方法了,因为这两个方法synchronized关键字合作使用。
此处我们需要使用Condition条件。
直接看代码:
// 进货
public void get() {
lock.lock();
try {
// 使用while防止虚假唤醒
while(product >= capacity) {
System.out.println("产品已满！");
try {
// 在一个参数版本中，中断和虚假的唤醒是可能的，这个方法应该总是在循环中使用:
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 除非product<capacity,否则不执行++product操作
System.out.println(Thread.currentThread().getName() + " : " + ++product);
condition.signalAll();
}finally {
lock.unlock();
}
}
// 卖货
public void sale() {
lock.lock();
try {
while (product <= 0) {
System.out.println("缺货！");
try {
// 在一个参数版本中，中断和虚假的唤醒是可能的，这个方法应该总是在循环中使用:
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 除非product>capacity,否则不执行--product操作
System.out.println(Thread.currentThread().getName() + " : " + --product);
condition.signalAll();
}finally {
lock.unlock();
}
}
我们定义了一个可重入锁 ReentrantLock lock, 通过lock.lock()来加锁，通过lock.unlock()来释放锁，让加锁范围更加灵活。
这里提一下Condition接口提供的方法：
- await()：方法会使当前线程等待，同时释放当前锁。当其他线程使用signal()或signalAll()方法时，线程会重新获得锁并继续执行。当线程被中断时，也能跳出等待。
- await(long time,TimeUnit unit)：方法会使线程等待，直到其他方法调用aignal()或者signalAll() 或者被中断，或者等待超过设置的时间
- awaitUninterruptibly()：方法与await()基本相同，但它并不会在等待的时候响应中断。
- signal()：唤醒一个等待的线程。如果有线程正在等待此条件，则选择一个线程进行唤醒。然后，该线程必须在从await返回之前重新获取锁。
- signalAll():唤醒所有等待的线程。如果有线程在这种情况下等待，那么它们将被唤醒。每个线程必须重新获取锁，然后才能从await返回。
最后
感谢你看到这里，文章有什么不足还请指正，觉得文章对你有帮助的话记得给我点个赞，每天都会分享java相关技术文章或行业资讯，欢迎大家关注和转发文章！