Java 生产者消费者模型的三种实现过程
生产者一边在生产,消费者一边消耗。当库存满的时候生产者暂停生产,直到有空位;当库存空的时候消费者暂停消费,直到有产品。
关键点:
- 生产者和消费者都是在不断生产和消费的,是同时并发的,不应该等满了再消费 / 空了再生产。
- 前提条件是生产速度和消费速度不同。
设计方法:
- 生产者和消费者分别使用一个线程模拟。将其抽象分别继承
Runnable接口
,生产和消费的行为放在Runnable
的run()方法
中. - 控制生产和消费的速度不同。可以使每次生产 / 每次消费 之间的间隔不同,达到 生产比消费快 或者 消费比生产快 的效果.
- 生产者拿到
仓库锁
,判断是否还有空位:
- 如果没有空位,就让自己先别再去试图抢占锁了(将自己放入等待池,
wait()
); - 如果有空位,就生产一个,然后唤醒(
notify()
)等待池中的线程(包括先前因为仓库空而wait()
掉的线程)来竞争锁。
- 消费者拿到
仓库锁
,判断是否还有可以消费的商品:
- 如果没有商品了,就让自己先别再去试图抢占锁了(将自己放入等待池,
wait()
); - 如果有空位,就消费一个,然后唤醒(
notify()
)等待池中的线程(包括先前因为仓库满而wait()
掉的线程)来竞争锁。
1.使用synchronized 关键字
使用wait
和notify
实现
public class ProducerAndConsumerModel {
public static void main(String[] args) {
list = new ArrayList<>();
new Thread(new Producer()).start();
new Thread(new Consumer()).start();
}
static int MAX_SIZE = 10;
static List<Integer> list;
private static class Producer implements Runnable{
@Override
public void run() {
while(true){
try {
Thread.sleep(250);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized(list){
if(list.size() >= MAX_SIZE){
try {
list.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
list.add(1);
System.out.println("生产了一个产品,现在有" + list.size() + "个产品");
list.notifyAll();
}
}
}
}
private static class Consumer implements Runnable{
@Override
public void run() {
while(true){
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (list){
if(list.size() == 0){
try {
list.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
list.remove(0);
System.out.println("消费了一个产品,现在还有" + list.size() + "个产品");
list.notifyAll();
}
}
}
注意:我们在这里使用的notifyAll()
这个方法,为什么不能用notify()
,也就是随便叫醒一个消费者呢?
答案是不可以,使用 notify()
是叫醒 LOCK 阻塞队列里面的任意一个线程,假如此时我们的临界区域已经满了,此时唤醒的是一个生产者线程,就会导致死锁,所以我们在这里采用的是notifyAll()
这个方法,意思就是唤醒阻塞队列里面的全部线程,这样某一个消费者就可以去取出临界区里面的产品,从而避免死锁的发生。
2.使用ReenterantLock
’
ReenterantLock
与synchronized
类似,synchronized
在语句块结束后会自动释放锁,ReenterantLock
需要unlock()
。ReenterantLock
还支持公平锁(等待最久的先拿锁)。
使用await
与singalAll
实现
public class ProducerAndConsumerModelReentrantLock {
public static void main(String[] args) {
list = new ArrayList<>();
ableToProducer = lock.newCondition();
ableToConsumer = lock.newCondition();
new Thread(new Producer()).start();
new Thread(new Consumer()).start();
}
static int MAX_SIZE = 10;
static List<Integer> list;
static ReentrantLock lock = new ReentrantLock(true);
static Condition ableToProducer,ableToConsumer;
private static class Producer implements Runnable{
@Override
public void run() {
while(true){
try {
Thread.sleep(250);
} catch (InterruptedException e) {
e.printStackTrace();
}
lock.lock();
try {
if(list.size() >= MAX_SIZE){
try {
ableToProducer.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
list.add(1);
System.out.println("生产了一个产品,现在有" + list.size() + "个产品");
ableToConsumer.signalAll();
}finally {
lock.unlock();
}
}
}
}
private static class Consumer implements Runnable{
@Override
public void run() {
while(true){
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
lock.lock();
try{
if(list.size() == 0){
try {
ableToConsumer.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
list.remove(0);
System.out.println("消费了一个产品,现在还有" + list.size() + "个产品");
ableToProducer.signalAll();
}finally {
lock.unlock();
}
}
}
}
}
与 synchronized
相比之下,一个 lock 我们可以生成多个condition
,换句话说synchronized
就像是只有一个 condition
的ReentrantLock
,所以 后者比前者更加的灵活,可以进行选择性的通知,但是也较为麻烦,因为每次都得手动地关闭锁,所以我们每次得尝试在finally
里面关闭锁。
3. 使用BlockingQueue
阻塞队列本身就是当队满的时候阻塞线程,故直接使用就可以实现生产者/消费者模型。
阻塞队列内部也是使用ReenterantLock
实现的。
public class ProducerAndConsumerModelUseBlockingQueue {
public static void main(String[] args) {
queue =new ArrayBlockingQueue<Integer>(MAX_SIZE);
new Thread(new Producer()).start();
new Thread(new Consumer()).start();
}
static BlockingQueue<Integer> queue;
static int MAX_SIZE = 10;
private static class Producer implements Runnable{
@Override
public void run() {
while(true){
try {
Thread.sleep(250);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
queue.put(1);
System.out.println("生产了一个产品,现在有"+queue.size()+"个产品。");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
private static class Consumer implements Runnable{
@Override
public void run() {
while(true) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
queue.take();
System.out.println("消费了一个产品,现在有"+queue.size()+"个产品。");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)