基于线程实现的生产者消费者模型(Object.wait(),Object.notify()方法)

需求背景

利用线程来模拟生产者和消费者模型

系统建模

这个系统涉及到三个角色，生产者，消费者，任务队列，三个角色之间的关系非常简单，生产者和消费者拥有一个任务队列的引用，生产者负责往队列中放置对象(id),消费者负责从队列中获取对象(id),其关联关系如下

方案1

因为是多线程操作，所以对任务的存取都要使用线程同步加锁机制，看一下我们的TaskQueue类，两个主方法都加了synchronized修饰，这就意味着，一个时间点只可能有一个线程对这个方法进行操作

TaskQueue类代码

[java] view plain copy

print?

1. package com.crazycoder2010.thread;
2. public class TaskQueue {
3. private int id;
4. public synchronized void put(int id){
5. this.id = id;
6. System.out.println("Put:"+id);
7. }
8. public synchronized int get(){
9. System.out.println("Got:"+this.id);
10. return this.id;
11. }
12. }

再来看一下生产者，这个主要不停的往TaskQueue中添加新对象，这里我们搞了个死循环，运行时不断的对当前数字做加一操作如下

Producer类代码

[java] view plain copy

print?

1. package com.crazycoder2010.thread;
2. public class Producer implements Runnable{
3. private TaskQueue taskQuery;
4. public Producer(TaskQueue taskQuery){
5. this.taskQuery = taskQuery;
6. new Thread(this).start();
7. }
8. @Override
9. public void run() {
10. int i = 0;
11. while(true){
12. taskQuery.put(i++);
13. }
14. }
15. }

消费者对象也是基于线程实现，在循环中不停的轮训TaskQuery.get()来获取当前对象

Consumer类

[java] view plain copy

print?

1. package com.crazycoder2010.thread;
2. public class Consumer implements Runnable {
3. private TaskQueue taskQuery;
4. public Consumer(TaskQueue taskQuery){
5. this.taskQuery = taskQuery;
6. new Thread(this).start();
7. }
8. @Override
9. public void run() {
10. while(true){
11. taskQuery.get();
12. }
13. }
14. }

运行一下，看看结果是否是我们所期望的那样

Launcher类代码

[java] view plain copy

print?

1. package com.crazycoder2010.thread;
2. public class Launcher {
3. public static void main(String[] args) {
4. TaskQueue taskQuery = new TaskQueue();
5. new Producer(taskQuery);
6. new Consumer(taskQuery);
7. System.out.println("Press Control-C to stop.");
8. }
9. }

输出结果：

...

Put:58

Put:59

Put:60

Put:61

Put:62

Put:63

Put:64

Got:64

Got:64

Got:64

Put:65

Put:66

...

问题出现在哪里呢？

从结果输出我们可以看出，生产者的对象放入顺序是按次序进行的，但是消费者读取对象的数序很奇怪，一段时间内读取同一个数值，这个是怎么造成的呢？

我们知道，当启动线程后线程什么时候被jvm所调度是不确定的，上面的结果在不同的机器上运行很可能得到不同的结果，当Producer线程被调度运行一段时间后，线程Consumer获取到运行资格，然后从TaskQueue中取对象，这个时候由于Producer处于等待被调度状态，所以TaskQueue中的id一直都是个固定值,所以这个时候Consumer获取到的一直都是Producer在被jvm设置为等待状态那一刻的值，运行一段时间，Producer又被jvm调度器调度，获取运行资格，这个时候id继续从上次暂定时的值开始累加，依次循环，然后就得到了上面的运行结果

方案2

这个方案里我们要对方案1做一些改造，当有Producer生产出一个id时，直到有ConSumer来将他拿走，然后再生产下一个id,Consumer也是类似，直到Producer生产出id后才来取，否则一直等待下去

解决：

Object类里有个wait()方法和notify()/notifyAll()，一直很神秘，先前没怎么用过，看了一下原来这个东东是与线程同步操作密接相关的，

比如我们在应用中如果要对某一个方法或某个代码段做线程同步控制，那么需要为这个方法添加synchronized修饰或者是synchronized(obj){},这个obj可以理解成我们实际生活中房间的一把锁，默认情况下，当一个线程拥有了一个方法或代码段的锁后，就可以进入房间(方法或代码块)任意干坏事，而其他的线程只能在门外等待直到当前线程执行完毕打开房间(释放锁)，而Object的wait和notify则是给这个锁提供了一些更先进的功能，这个锁可以自己开锁wait()(让同步方法或代码块暂时放弃占用的锁)，进而让别的线程有机会进入运行，而当实际成熟时(满足运行的条件)又可以把自身锁住notify(),进而又继续进入上次被暂停的操作

改造后的TaskQueue2

[java] view plain copy

print?

1. package com.crazycoder2010.thread;
2. public class TaskQuery2 {
3. private int id;
4. private boolean valueSet;
5. public synchronized int get(){
6. if(!valueSet){
7. try {
8. wait();
9. } catch (InterruptedException e) {
10. e.printStackTrace();
11. }
12. }
13. valueSet = false;
14. notify();
15. System.out.println("Got:"+this.id);
16. return this.id;
17. }
18. public synchronized void put(int id){
19. if(valueSet){
20. try {
21. wait();
22. } catch (InterruptedException e) {
23. e.printStackTrace();
24. }
25. }
26. this.id = id;
27. valueSet = true;
28. System.out.println("Put:"+id);
29. notify();
30. }
31. }

Producer,Consumer和Launcher类的代码不改动

运行结果如下

Put:0

Got:0

Put:1

Got:1

Put:2

Got:2

Put:3

Got:3

Put:4

Got:4

Put:5

Got:5

Put:6

Got:6