生产者消费者模型

什么是生产者消费者

用一个不是特别恰当的例子做比喻:鸡下蛋,人吃鸡蛋,鸡每天下10个蛋,但是人每天只能吃2个蛋,吃不及蛋就会造成阻塞。生产者消费者模型呢,给中间放了个篮子,鸡下蛋放进篮子,人吃蛋从篮子里拿,将人与鸡的强耦合解开,变成了弱耦合。

生产者消费者模型需要关注的几个点:

  • 生产者生产的时候消费者不能消费
  • 消费者消费的时候生产者不能生产
  • 缓冲区空时消费者不能消费
  • 缓冲区满时生产者不能生产

为什么要用生产者消费者模型?优点在哪里?

  • 解耦 。因为多了一个缓冲区,所以生产者和消费者并不直接相互调用,这一点很容易想到,这样生产者和消费者的代码发生变化,都不会对对方产生影响,这样其实就把生产者和消费者之间的强耦合解开,变为了生产者和缓冲区/消费者和缓冲区之间的弱耦合
  • 通过平衡生产者和消费者的处理能力来提高整体处理数据的速度,这是生产者/消费者模型最重要的一个优点。如果消费者直接从生产者这里拿数据,如果生产者生产的速度很慢,但消费者消费的速度很快,那消费者就得占用CPU的时间片白白等在那边。有了生产者/消费者模型,生产者和消费者就是两个独立的并发体,生产者把生产出来的数据往缓冲区一丢就好了,不必管消费者;消费者也是,从缓冲区去拿数据就好了,也不必管生产者,缓冲区满了就不生产,缓冲区空了就不消费,使生产者/消费者的处理能力达到一个动态的平衡

实现生产者消费者模型的两个简单Demo


利用wait()/notify()实现生产者/消费者模型

既然生产者/消费者模型有一个缓冲区,那么我们就自己做一个缓冲区,生产者和消费者的通信都是通过这个缓冲区的。value为””表示缓冲区空,value不为””表示缓冲区满:

1
2
3
4
public class ValueObject
{
public static String value = "";
}

接下来就是一个生产者了,如果缓冲区满了的,那么就wait(),不再生产了,等待消费者消费完通知;如果缓冲区是空的,那么就生产数据到缓冲区中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package com.myhexin.ioc.producerConsumer;

public class Producer {
private Object lock;

public Producer(Object lock){
this.lock = lock;
}

public void setValue(){
try {
synchronized (lock){
if (!ValueObject.value.equals("")){
lock.wait();
}
String value = System.currentTimeMillis() + "_" + System.nanoTime();
System.out.println("Set的值为:" + value);
ValueObject.value = value;
lock.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

消费者类似,如果缓冲区是空的,那么就不再消费,wait()等待,等待生产者生产完通知;如果缓冲区不是空的,那么就去拿数据:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
package com.myhexin.ioc.producerConsumer;

public class Customer {
private Object lock;

public Customer(Object lock){
this.lock = lock;
}

public void getValue(){
try {
synchronized (lock){
if (ValueObject.value.equals("")){
lock.wait();
}
System.out.println("Get的值为:" + ValueObject.value);
ValueObject.value = "";
lock.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

写个主函数,开两个线程调用Producer里面的getValue()方法和Customer()里面的setValue()方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package com.myhexin.ioc.producerConsumer;

public class Hello {
public static void main(String[] args) {
Object lock = new Object();
final Producer producer = new Producer(lock);
final Customer customer = new Customer(lock);

Runnable producerRunnable = new Runnable() {
@Override
public void run() {
while (true){
producer.setValue();
}
}
};

Runnable customerRunnable = new Runnable() {
@Override
public void run() {
while (true){
customer.getValue();
}
}
};

Thread p = new Thread(producerRunnable);
Thread c = new Thread(customerRunnable);
p.start();
c.start();
}
}

看一下运行结果:

1
2
3
4
5
6
7
8
...
Set的值是:1444025677743_162366875965845
Get的值是:1444025677743_162366875965845
Set的值是:1444025677743_162366875983541
Get的值是:1444025677743_162366875983541
Set的值是:1444025677743_162366876004776
Get的值是:1444025677743_162366876004776
...

生产数据和消费数据一定是成对出现的,生产一个消费一个,满了不生产,空了不消费,生产者不能无限生产,消费者也不能无限消费,符合生产者/消费者模型。生产者速度快,就不占用CPU时间片,等着消费者消费完通知它继续生产,这块时间片可以用来给其他线程用。


利用await()/signal()实现生产者和消费者模型

一样,先定义一个缓冲区:

1
2
3
4
public class ValueObject
{
public static String value = "";
}

换种写法,生产和消费方法放在一个类里面:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package com.myhexin.ioc.pc2;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ThreadDomain extends ReentrantLock {

private Condition condition = newCondition();

public void set(){
try {
lock();
while (!"".equals(ValueObject.value))
condition.await();
ValueObject.value = "123";
System.out.println(Thread.currentThread().getName() + "生产了:" + ValueObject.value);
condition.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
unlock();
}
}
public void get(){
try {
lock();
while ("".equals(ValueObject.value))
condition.await();
System.out.println(Thread.currentThread().getName() + "消费了:" + ValueObject.value);
ValueObject.value = "";
condition.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
unlock();
}
}

}

同样的,开两个线程,一个线程调用set()方法生产,另一个线程调用get()方法消费:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
package com.myhexin.ioc.pc2;

public class Hello {
public static void main(String[] args) {
final ThreadDomain td = new ThreadDomain();
Runnable pRunnable = new Runnable() {
@Override
public void run() {
while (true)
td.set();
}
};

Runnable cRunnable = new Runnable() {
@Override
public void run() {
while (true)
td.get();
}
};

Thread p = new Thread(pRunnable);
p.setName("P");
Thread c = new Thread(cRunnable);
c.setName("C");
p.start();
c.start();
}
}

和wait()/notify()机制的实现效果一样,同样符合生产者/消费者模型.

小心假死

生产者/消费者模型最终达到的目的是平衡生产者和消费者的处理能力,达到这个目的的过程中,并不要求只有一个生产者和一个消费者。可以多个生产者对应多个消费者,可以一个生产者对应一个消费者,可以多个生产者对应一个消费者。

假死就发生在上面三种场景下。理论分析就能说明问题,所以就不写代码了。代码要写也很简单,上面的两个例子随便修改一个,开一个生产者线程/多个消费者线程、开多个生产者线程/消费者线程、开多个生产者线程/多个消费者线程都可以。假死指的是全部线程都进入了WAITING状态,那么程序就不再执行任何业务功能了,整个项目呈现停滞状态。

比方说有生产者A和生产者B,缓冲区由于空了,消费者处于WAITING。生产者B处于WAITING,生产者A被消费者通知生产,生产者A生产出来的产品本应该通知消费者,结果通知了生产者B,生产者B被唤醒,发现缓冲区满了,于是继续WAITING。至此,两个生产者线程处于WAITING,消费者处于WAITING,系统假死。

上面的分析可以看出,假死出现的原因是因为notify的是同类,所以非单生产者/单消费者的场景,可以采取两种方法解决 这个问题:

1、synchronized用notifyAll()唤醒所有线程、ReentrantLock用signalAll()唤醒所有线程

2、用ReentrantLock定义两个Condition,一个表示生产者的Condition,一个表示消费者的Condition,唤醒的时候调用相应的Condition的signal()方法就可以了

本文转自五月的仓颉大神,原文链接:https://www.cnblogs.com/xrq730/p/4855663.html

0%