Producer Consumer 模式

Producer Consumer 模式是指一个负责生产，一个负责消费。

核心是生产者安全地将数据交给消费者。

实际案例

类信息概览：

类名 说明
Main.java 方法的总入口
ConsumerCakeThread.java 消费蛋糕的线程
Table.java 放置蛋糕的桌子
BlockingQueueTable.java BlockingQueue 实现
ProducerCakeThread.java 生产蛋糕的线程

定义

  • ConsumerCakeThread.java
  [java]
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
package com.github.houbb.thread.learn.easy.learn.producer.consumer;

import java.util.Random;

public class ConsumerCakeThread extends Thread {

    private String name;

    private final Table table;

    public ConsumerCakeThread(String name, Table table) {
        super(name);
        this.table = table;
    }

    @Override
    public void run() {
        Random random = new Random(1000L);
        try {
            while(true) {
                Thread.sleep(random.nextInt(1000));
                String cake = table.take();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
  • Table.java
  [java]
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
package com.github.houbb.thread.learn.easy.learn.producer.consumer;

/**
 * @see BlockingQueueTable 可以被这个替换
 */
public class Table {

    /**
     * 存放蛋糕的数组
     */
    private String[] cakeArray;

    /**
     * 头
     */
    private int head;

    /**
     * 尾巴
     */
    private int tail;

    private int count;

    private final int size;


    public Table(int size) {
        this.size = size;
        cakeArray = new String[size];
        this.head = 0;
        this.tail = 0;
        this.count = 0;
    }

    public synchronized void put(final String cakeName) throws InterruptedException {
        while(count >= size) {
            wait();
        }

        cakeArray[tail] = cakeName;
        count++;
        tail = (tail + 1) % size;
        System.out.println(Thread.currentThread().getName() + " put cake " + cakeName);

        notifyAll();
    }

    public synchronized String take() throws InterruptedException {
        while (count <= 0) {
            wait();
        }

        String result = cakeArray[head];
        head = (head + 1) % size;
        count--;

        System.out.println(Thread.currentThread().getName() + " take cake " + result);
        notifyAll();

        return result;
    }

}
  • BlockingQueueTable.java
  [java]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
package com.github.houbb.thread.learn.easy.learn.producer.consumer;

import java.util.concurrent.ArrayBlockingQueue;

/**
 * 使用 queue
 */
public class BlockingQueueTable extends ArrayBlockingQueue<String> {

    public BlockingQueueTable(int size) {
        super(size);
    }

    public void put(final String cakeName) throws InterruptedException {
        super.put(cakeName);
    }

    public String take() throws InterruptedException {
        return super.take();
    }

}
  • ProducerCakeThread.java
  [java]
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
package com.github.houbb.thread.learn.easy.learn.producer.consumer;

import java.util.Random;

public class ProducerCakeThread extends Thread {

    private final Table table;

    private static int id = 0;

    public ProducerCakeThread(String name, Table table) {
        super(name);
        this.table = table;
    }

    @Override
    public void run() {
        Random random = new Random(1000L);
        try {
            while(true) {
                String cakeName = getName()+"-"+genId();
                Thread.sleep(random.nextInt(1000));
                table.put(cakeName);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static synchronized int genId() {
        return id++;
    }

}

测试

  • Main.java
  [java]
1
2
3
4
5
6
7
8
9
10
11
12
package com.github.houbb.thread.learn.easy.learn.producer.consumer;

public class Main {

    public static void main(String[] args) {
        Table table = new Table(3);
        new ConsumerCakeThread("ConsumerCake", table).start();
        new ProducerCakeThread("ProducerCake", table).start();
    }

}
  • 测试结果
  [plaintext]
1
2
3
4
5
6
7
8
9
10
11
12
ProducerCake put cake ProducerCake-0
ConsumerCake take cake ProducerCake-0
ProducerCake put cake ProducerCake-1
ConsumerCake take cake ProducerCake-1
ProducerCake put cake ProducerCake-2
ConsumerCake take cake ProducerCake-2
ProducerCake put cake ProducerCake-3
ConsumerCake take cake ProducerCake-3
ProducerCake put cake ProducerCake-4
ConsumerCake take cake ProducerCake-4
ProducerCake put cake ProducerCake-5
ConsumerCake take cake ProducerCake-5

实现方式

UML & Code

UML

UML 图示如下

Producer Consumer

Code

代码地址

Producer Consumer

系列导航

多线程系列导航