开发笔记:Java并发编程生产者与消费者模式介绍

package com.yuanfy.jmm.threads;
import com.yuanfy.util.SleepUtils;
import java.util.concurrent.TimeUnit;
public class Factory {
// 当前库存大小
private int size;
// 库存容量（最大库存值）
private int capacity;
public Factory(int capacity) {
this.capacity = capacity;
}
public synchronized void produce(int num) {
try {
System.out.println("+++++生产者【" + Thread.currentThread().getName()
+ "】, 他的任务是生产" + num + "件产品.");
// 当生产完成就停止
while (num > 0) {
// 如果当前库存大小大于或等于库存容量值了，则停止生产等待消费。
if (size >= capacity) {
System.out.println("+++++" + Thread.currentThread().getName() +
"检测库存已满，停止生产等待消费...");
// 等待消费
wait();
System.out.println("+++++" + Thread.currentThread().getName() + "开始生产...");
}
// 否则继续生产
int inc = (num + size) > capacity ? (capacity - size) : num;
num -= inc;
size += inc;
SleepUtils.second(1);
System.out.println("+++++" + Thread.currentThread().getName() + " 生产了" + inc + "件，当前库存有" + size + "件.");
// 生产后唤醒消费者
notify();
}
System.out.println("+++++生产者【" + Thread.currentThread().getName()
+ "】 生产结束.");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public synchronized void consume(int num) {
try {
System.out.println("-----消费者【" + Thread.currentThread().getName()
+ "】, 他需要消费" + num + "件产品.");
// 当消费完成则停止
while (num > 0) {
// 如果当前库存大小小于等于0，则停止消费等待生产。
if (size <= 0) {
System.out.println("-----" + Thread.currentThread().getName() + " 检测库存已空，停止消费等待生产...");
// 等待生产
wait();
System.out.println("-----" + Thread.currentThread().getName() + " 开始消费...");
}
// 否则继续消费
int dec = (size - num) > 0 ? num : size;
num -= dec;
size -= dec;
SleepUtils.second(1);
System.out.println("-----" + Thread.currentThread().getName() + " 消费了" + dec + "件，当前有" + size + "件.");
// 消费后唤醒生产者继续生产
notify();
}
System.out.println("-----消费者【" + Thread.currentThread().getName()
+ "】 消费结束.");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

package com.yuanfy.jmm.threads;
/**
* 生产线程
*/
class Produce {
private Factory factory;
public Produce(Factory factory) {
this.factory = factory;
}
public void produce(String name, final int num) {
new Thread(new Runnable() {
@Override
public void run() {
factory.produce(num);
}
}, name).start();
}
}
/**
* 消费线程
*/
class Consume {
private Factory factory;
public Consume(Factory factory) {
this.factory = factory;
}
public void consume(String name, final int num) {
new Thread(new Runnable() {
@Override
public void run() {
factory.consume(num);
}
}, name).start();
}
}
public class ProduceConsumeDemo {
public static void main(String[] args) {
Factory f = new Factory(500);

Consume consume = new Consume(f);
consume.consume("消费线程",600);
Produce produce = new Produce(f);
produce.produce("生产线程",800);
}
}

public class Factory {
// 当前大小
private int size;
// 总容量
private int capacity;
private Lock lock;
// 已满的条件
private Condition fullCondition;
// 已空的条件
private Condition emptyCondition;
public Factory(int capacity) {
this.capacity = capacity;
lock = new ReentrantLock();
fullCondition = lock.newCondition();
emptyCondition = lock.newCondition();
}
public void produce(int no) {
lock.lock();
try {
while (no > 0) {
while (size >= capacity) {
System.out.println(Thread.currentThread().getName() + " 报告仓库已满，等待快递员取件...");
fullCondition.await();
System.out.println(Thread.currentThread().getName() + " 报告开始进货...");
}
int inc = (no + size) > capacity ? (capacity - size) : no;
no -= inc;
size += inc;
System.out.println(Thread.currentThread().getName() +
" 报告进货了： " + inc + "件, 当前库存数： " + size);
emptyCondition.signal();
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void consume(int no) {
lock.lock();
try {
while (no > 0) {
while (size <= 0) {
System.out.println(Thread.currentThread().getName() + " 报告仓库已空，等待仓库管理员进货");
emptyCondition.await();
System.out.println(Thread.currentThread().getName() + " 报告开始取件...");
}
int dec = (size - no) > 0 ? no : size;
no -= dec;
size -= dec;
System.out.println(Thread.currentThread().getName() +
" 报告取件： " + dec + ", 当前库存数： " + size);
fullCondition.signal();
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}