JAVA深化篇_33——线程并发协作(生产者/消费者模式)

线程并发协作(生产者/消费者模式)

多线程环境下，我们经常需要多个线程的并发和协作。这个时候，就需要了解一个重要的多线程并发协作模型“生产者/消费者模式”。

角色介绍

• 什么是生产者？

  生产者指的是负责生产数据的模块（这里模块可能是：方法、对象、线程、进程）。

• 什么是消费者？

  消费者指的是负责处理数据的模块（这里模块可能是：方法、对象、线程、进程）。

• 什么是缓冲区？

  消费者不能直接使用生产者的数据，它们之间有个“缓冲区”。生产者将生产好的数据放入“缓冲区”，消费者从“缓冲区”拿要处理的数据。

缓冲区是实现并发的核心，缓冲区的设置有两个好处：

1. 实现线程的并发协作

   有了缓冲区以后，生产者线程只需要往缓冲区里面放置数据，而不需要管消费者消费的情况；同样，消费者只需要从缓冲区拿数据处理即可，也不需要管生产者生产的情况。 这样，就从逻辑上实现了“生产者线程”和“消费者线程”的分离，解除了生产者与消费者之间的耦合。

2. 解决忙闲不均，提高效率

   生产者生产数据慢时，缓冲区仍有数据，不影响消费者消费；消费者处理数据慢时，生产者仍然可以继续往缓冲区里面放置数据 。

实现生产者与消费者模式

创建缓冲区

/**
 * 定义馒头类
 */
class ManTou{
  private int id;
  public ManTou(int id){
    this.id = id;
   }
  public int getId(){
    return this.id;
   }
}


/**
 * 定义缓冲区类
 */
class SyncStack{
  //定义存放馒头的盒子
  private ManTou[] mt = new ManTou[10];
  //定义操作盒子的索引
  private int index;


  /**
   * 放馒头
   */
  public synchronized void push(ManTou manTou){
    //判断盒子是否已满
    while(this.index == this.mt.length){
      try {
        /**
         * 语法：wait(),该方法必须要在synchronized块中调用。
         * wait执行后，线程会将持有的对象锁释放，并进入阻塞状态，
         * 其他需要该对象锁的线程就可以继续运行了。
         */
        this.wait();
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
     }
    //唤醒取馒头的线程
    /**
     * 语法：该方法必须要在synchronized块中调用。
     * 该方法会唤醒处于等待状态队列中的一个线程。
     */
    this.notify();
    this.mt[this.index] = manTou;
    this.index++;
   }
  /**
   * 取馒头
   */
  public synchronized ManTou pop(){
    while(this.index == 0){
      try {
        /**
         * 语法：wait(),该方法必须要在synchronized块中调用。
         * wait执行后，线程会将持有的对象锁释放，并进入阻塞状态，
         * 其他需要该对象锁的线程就可以继续运行了。
         */
        this.wait();
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
     }
    this.notify();
    this.index--;
    return this.mt[this.index];
   }
}


public class TestProduceThread {
  public static void main(String[] args) {
    
   }
}

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

创建生产者消费者线程

/**
 * 定义馒头类
 */
class ManTou{
  private int id;
  public ManTou(int id){
    this.id = id;
   }
  public int getId(){
    return this.id;
   }
}


/**
 * 定义缓冲区类
 */
class SyncStack{
  //定义存放馒头的盒子
  private ManTou[] mt = new ManTou[10];
  //定义操作盒子的索引
  private int index;


  /**
   * 放馒头
   */
  public synchronized void push(ManTou manTou){
    //判断盒子是否已满
    while(this.index == this.mt.length){
      try {
        /**
         * 语法：wait(),该方法必须要在synchronized块中调用。
         * wait执行后，线程会将持有的对象锁释放，并进入阻塞状态，
         * 其他需要该对象锁的线程就可以继续运行了。
         */
        this.wait();
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
     }
    //唤醒取馒头的线程
    /**
     * 语法：该方法必须要在synchronized块中调用。
     * 该方法会唤醒处于等待状态队列中的一个线程。
     */
    this.notify();
    this.mt[this.index] = manTou;
    this.index++;
   }
  /**
   * 取馒头
   */
  public synchronized ManTou pop(){
    while(this.index == 0){
      try {
        /**
         * 语法：wait(),该方法必须要在synchronized块中调用。
         * wait执行后，线程会将持有的对象锁释放，并进入阻塞状态，
         * 其他需要该对象锁的线程就可以继续运行了。
         */
        this.wait();
       } catch (InterruptedException e) {
        e.printStackTrace();
       }
     }
    this.notify();
    this.index--;
    return this.mt[this.index];
   }
}


/**
 * 定义生产者线程类
 */
class ShengChan extends Thread{
  private SyncStack ss;
  public ShengChan(SyncStack ss){
    this.ss = ss;
   }
  @Override
  public void run() {
    for(int i=0;i<10;i++){
      System.out.println("生产馒头："+i);
      ManTou manTou = new ManTou(i);
      this.ss.push(manTou);
    }
   }
}


/**
 * 定义消费者线程类
 */
class XiaoFei extends Thread{
  private SyncStack ss;
  public XiaoFei(SyncStack ss){
    this.ss = ss;
   }
  @Override
  public void run() {
    for(int i=0;i<10;i++){
      ManTou manTou = this.ss.pop();
      System.out.println("消费馒头："+i);
     }
   }
}
public class ProduceThread {
  public static void main(String[] args) {
    SyncStack ss = new SyncStack();
    new ShengChan(ss).start();
    new XiaoFei(ss).start();
   }
}

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

线程并发协作总结

线程并发协作（也叫线程通信）

生产者消费者模式：

1. 生产者和消费者共享同一个资源，并且生产者和消费者之间相互依赖，互为条件。

2. 对于生产者，没有生产产品之前，消费者要进入等待状态。而生产了产品之后，又需要马上通知消费者消费。

3. 对于消费者，在消费之后，要通知生产者已经消费结束，需要继续生产新产品以供消费。

4. 在生产者消费者问题中，仅有synchronized是不够的。synchronized可阻止并发更新同一个共享资源，实现了同步但是synchronized不能用来实现不同线程之间的消息传递（通信）。

5. 那线程是通过哪些方法来进行消息传递（通信）的呢？见如下总结：

   方法名	作 用
   final void wait()	表示线程一直等待，直到得到其它线程通知
   void wait(long timeout)	线程等待指定毫秒参数的时间
   final void wait(long timeout,int nanos)	线程等待指定毫秒、微秒的时间
   final void notify()	唤醒一个处于等待状态的线程
   final void notifyAll()	唤醒同一个对象上所有调用wait()方法的线程，优先级别高的线程优先运行

6. 以上方法均是java.lang.Object类的方法；

都只能在同步方法或者同步代码块中使用，否则会抛出异常。

建议

在实际开发中，尤其是“架构设计”中，会大量使用这个模式。 对于初学者了解即可，如果晋升到中高级开发人员，这就是必须掌握的内容。