Java中的生产者消费者问题
一、生产者消费者问题
概念
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
321原则
优点:
解耦–生产者。消费者之间不直接通信，降低了耦合度。
支持并发、支持忙闲不均
图解
二、Synchronized 和wait/notify实现
/**
* 线程之间的通信问题：生产者消费者问题
* 线程交替执行，A B 操作同一个变量
*/
public class A {
public static void main(String[] args) {
Data data = new Data();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"A").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"B").start();
}
}
// 等待 业务 通知
class Data{ //资源类
private int num = 0;
public synchronized void increment() throws InterruptedException {
if(num !=0){
this.wait(); // 等待消费者消费
}
num++; // 消费池里没有资源了，生产者生产资源并通知消费者消费
System.out.println("生产者" + Thread.currentThread().getName() + "生产后=> num=" + num);
notifyAll();
}
public synchronized void decrement() throws InterruptedException {
if(num ==0){
this.wait(); // 等待生产者生产
}
// 消费池里还有资源没消费，消费者者消费资源后并通知生产者生产
num--;
System.out.println("消费者" + Thread.currentThread().getName() + "消费后=> num=" + num);
notifyAll();
}
}
问题：synchronized关键字，在多个线程时是否还能适用？
在上面的基础上，多添加两个线程，A B C D ，出现混乱，存在虚假唤醒的可能。
/**
* 线程之间的通信问题：生产者消费者问题
* 线程交替执行，A B 操作同一个变量
*/
public class A {
public static void main(String[] args) {
Data data = new Data();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"A").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"B").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"C").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"D").start();
}
}
// 等待 业务 通知
class Data{ //资源类
private int num = 0;
public synchronized void increment() throws InterruptedException {
if(num !=0){
this.wait(); // 等待消费者消费
}
num++; // 消费池里没有资源了，生产者生产资源并通知消费者消费
System.out.println("生产者" + Thread.currentThread().getName() + "生产后=> num=" + num);
notifyAll();
}
public synchronized void decrement() throws InterruptedException {
if(num ==0){
this.wait(); // 等待生产者生产
}
// 消费池里还有资源没消费，消费者者消费资源后并通知生产者生产
num--;
System.out.println("消费者" + Thread.currentThread().getName() + "消费后=> num=" + num);
notifyAll();
}
}
按照官方文档的提示，解决该问题： 将资源里面sychronized方法里面的if判断改为where判断语句。
三、Lock 和 await/signal实现
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 线程之间的通信问题：生产者消费者问题
* 线程交替执行，A B 操作同一个变量
*/
public class B {
public static void main(String[] args) {
Data2 data2 = new Data2();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data2.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"A").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data2.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"B").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data2.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"C").start();
new Thread(()->{
for (int i = 0; i < 10 ; i++) {
try {
data2.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"D").start();
}
}
// 等待 业务 通知
class Data2{ //资源类
private int num = 0;
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
public void increment() throws InterruptedException {
lock.lock(); // 加锁
try{ // 业务代码
while (num !=0){
condition.await();
}
num++; // 消费池里没有资源了，生产者生产资源并通知消费者消费
System.out.println("生产者" + Thread.currentThread().getName() + "生产后=> num=" + num);
condition.signalAll();
}catch (Exception e){
e.printStackTrace();
}finally {
lock.unlock(); //解锁
}
}
public void decrement() throws InterruptedException {
lock.lock(); // 加锁
try{ // 业务代码
while (num ==0){
condition.await();
}
num--; // 消费池里没有资源了，生产者生产资源并通知消费者消费
System.out.println("消费者" + Thread.currentThread().getName() + "消费后=> num=" + num);
condition.signalAll();
}catch (Exception e){
e.printStackTrace();
}finally {
lock.unlock(); //解锁
}
}
}
通过上面的结果，我们发现，线程的允许顺序是随机的，不满足有序执行，因此使用Condition实现精准通知和唤醒线程
package edu.cqupt.pc;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class C {
public static void main(String[] args) {
Data3 data3 = new Data3();
new Thread(()->{for (int i = 0; i <10 ; i++) {data3.printA();}},"A").start();
new Thread(()->{for (int i = 0; i <10 ; i++) {data3.printB();}},"B").start();
new Thread(()->{for (int i = 0; i <10 ; i++) {data3.printC();}},"C").start();
}
}
class Data3{
private Lock lock = new ReentrantLock();
private Condition condition1 = lock.newCondition();
private Condition condition2 = lock.newCondition();
private Condition condition3 = lock.newCondition();
private int num =1;
public void printA(){
lock.lock();
try {
// 业务、 判断-> 执行->通知
while(num !=1){
// 等待
condition1.await();
}
System.out.println(Thread.currentThread().getName()+"AAA");
num = 2;
// 精准唤醒B
condition2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printB(){
lock.lock();
try {
if(num !=2){
//等待
condition2.await();
}
System.out.println(Thread.currentThread().getName()+"BBB");
num = 3;
// 精准唤醒C
condition3.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printC(){
lock.lock();
try {
while (num !=3){
condition3.await();
}
System.out.println(Thread.currentThread().getName()+"CCC");
num = 1;
// 精准唤醒A
condition1.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
在上面的基础上，实现了精准的唤醒线程，让线程依次执行。
参考: 狂神说Java的JUC部分 https://www.bilibili.com/video/BV1B7411L7tE?p=9&spm_id_from=pageDriver