生产者消费者模型具体来讲，就是在一个系统中，存在生产者和消费者两种角色，他们通过内存缓冲区进行通信（解耦），生产者将消费者需要的资源生产出来放到缓冲区，消费者把从缓冲区把资源拿走消费。
在这个模型中，最关键就是内存缓冲区为空的时候消费者必须等待，而内存缓冲区满的时候，生产者必须等待。其他时候就是一边在生产一边在消费。值得注意的是多线程对内存缓冲区的操作时必须保证线程安全，所以需要设计锁的策略。
◆
使用wait和notify实现生产这消费者
◆
我们在Hello，Thread一文中提到了wait和notify来实现等待通知的功能，本篇文章则继续使用它们实现一个生产者、消费者模型。
首先我们定义一个资源的类，资源类中初始时什么都没有，最多允许存放10个资源。
当生产者调用add方法时，i+1，即代表生产出了一件资源。当生产了一个资源以后就使用notifyAll通知所有等待在此资源文件的线程。如果当资源达到10个后则所有的生产者线程进入等待状态，等待消费者线程唤醒。
当消费者调用remove方法时，i-1，即代表消费了一件资源。当消费了一个资源以后就使用notifyAll通知所有等待在此资源文件的线程。如果当资源达到0个后则所有的消费者线程进入等待状态，等待生产者线程唤醒。
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
|
public class WaitNotifyResouce {
private int i=0;
private int size=10;
public synchronized void add(){
if(i<size){
i++;
System.out.println(Thread.currentThread().getName()+"号线程生产一件资源,当前资源"+i+"个");
notifyAll();
}else {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public synchronized void remove(){
if(i>0){
i--;
System.out.println(Thread.currentThread().getName()+"号线程拿走了一件资源,当前资源"+i+"个");
notifyAll();
}else {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
|
接下来我们创建3个生产者线程、2个消费者线程持续对资源进行生产和消费。
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
40
41
42
43
|
public class WaitNotifyProducerConsumerDemo {
static WaitNotifyResouce waitNotifyResouce = new WaitNotifyResouce();
static class ProducerThreadDemo extends Thread {
@Override
public void run() {
while (true) {
try {
sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
waitNotifyResouce.add();
}
}
}
static class ConsumerThreadDemo extends Thread {
@Override
public void run() {
while (true) {
try {
sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
waitNotifyResouce.remove();
}
}
}
public static void main(String[] args) throws InterruptedException {
Thread p1 = new Thread(new ProducerThreadDemo(), "生产者p1");
Thread p2 = new Thread(new ProducerThreadDemo(), "生产者p2");
Thread p3 = new Thread(new ProducerThreadDemo(), "生产者p3");
p1.start();
p2.start();
p3.start();
Thread c1 = new Thread(new ConsumerThreadDemo(), "消费者c1");
Thread c2 = new Thread(new ConsumerThreadDemo(), "消费者c2");
c1.start();
c2.start();
}
}
|
接下来程序打印的结果就像预想中一样了：
1
2
3
4
5
6
7
8
9
|
生产者p1号线程生产一件资源,当前资源1个
生产者p2号线程生产一件资源,当前资源2个
生产者p3号线程生产一件资源,当前资源3个
消费者c1号线程拿走了一件资源,当前资源2个
消费者c2号线程拿走了一件资源,当前资源1个
生产者p1号线程生产一件资源,当前资源2个
生产者p3号线程生产一件资源,当前资源3个
生产者p2号线程生产一件资源,当前资源4个
。。。
|
◆
使用Condition实现生产者消费者模型
◆
在文章：浅谈Java中的锁：Synchronized、重入锁、读写锁 中，我们了解了 Lock和Condition，现在我们使用它们配合实现一个生产者消费者模型
首先同样创建一个资源文件，此资源文件所有的操作跟上方的资源文件是一样的，只不过使用Lock和Condition的组合代替了synchronize。
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
40
41
42
43
44
45
46
|
public class LockConditionResouce {
private int i = 0;
private int size = 10;
private Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
public void add() {
lock.lock();
try {
if (i < size) {
i++;
System.out.println(Thread.currentThread().getName() + "号线程生产一件资源,当前资源" + i + "个");
condition.signalAll();
} else {
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} finally {
lock.unlock();
}
}
public void remove() {
lock.lock();
try {
if (i > 0) {
i--;
System.out.println(Thread.currentThread().getName() + "号线程拿走了一件资源,当前资源" + i + "个");
condition.signalAll();
} else {
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} finally {
lock.unlock();
}
}
}
|
接下来使用生产者消费者线程操作资源：
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
40
41
42
|
public class LockConditionProducerConsumerDemo {
static LockConditionResouce lockConditionResouce = new LockConditionResouce();
static class ProducerThreadDemo extends Thread {
@Override
public void run() {
while (true) {
try {
sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
lockConditionResouce.add();
}
}
}
static class ConsumerThreadDemo extends Thread {
@Override
public void run() {
while (true) {
try {
sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
lockConditionResouce.remove();
}
}
}
public static void main(String[] args) throws InterruptedException {
Thread p1 = new Thread(new ProducerThreadDemo(), "生产者p1");
Thread p2 = new Thread(new ProducerThreadDemo(), "生产者p2");
Thread p3 = new Thread(new ProducerThreadDemo(), "生产者p3");
p1.start();
p2.start();
p3.start();
Thread c1 = new Thread(new ConsumerThreadDemo(), "消费者c1");
Thread c2 = new Thread(new ConsumerThreadDemo(), "消费者c2");
c1.start();
c2.start();
}
}
|