【图解Java多线程设计模式】Producer-Consumer模式

Posted on | In

生产者安全地将数据交给消费者。Producer-Consumer模式在生产者和消费者之间加入了一个“桥梁角色”。该桥梁角色用于消除线程间处理速度的差异。

示例

有3位糕点师制作蛋糕并将其放到桌子上，然后有3位客人来吃这些蛋糕。

时序图

FZ98BR.md.png

Main.java

1
2
3
4
5
6
7
8
9
10
11
12
public class Main {

    public static void main(String[] args) {
        Table table = new Table(3);
        new MakerThread("MakerThread-1", table, 31415).start();
        new MakerThread("MakerThread-2", table, 92653).start();
        new MakerThread("MakerThread-3", table, 58979).start();
        new EaterThread("EaterThread-1", table, 32384).start();
        new EaterThread("EaterThread-2", table, 62643).start();
        new EaterThread("EaterThread-3", table, 38327).start();
    }
}

MakerThread.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
import java.util.Random;

public class MakerThread extends Thread {
    private final Table table;
    private static int id = 0;
    private final Random random;
    
    public MakerThread(String name, Table table, long seed) {
        super(name);
        this.table = table;
        this.random = new Random(seed);
    }

    public void run() {
        try {
            while (true) {
                Thread.sleep(random.nextInt(1000));
                String cake = "[ Cake No." + nextId() + " by " + getName() + " ]";
                table.put(cake);
            }
        } catch (InterruptedException e) {
        }
    }

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

EaterThread.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import java.util.Random;

public class EaterThread extends Thread {
    private final Table table;
    private final Random random;
    
    public EaterThread(String name, Table table, long seed) {
        super(name);
        this.table = table;
        this.random = new Random(seed);
    }

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

Table.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
public class Table {
    private final String[] buffer;
    private int head;
    private int tail;
    private int count;

    public Table(int count) {
        this.buffer = new String[count];
        this.head = 0;
        this.tail = 0;
        this.count = 0;
    }
    
    public synchronized void put(String cake) throws InterruptedException {
        while (count >= buffer.length) {
            wait();
        }

        System.out.println(Thread.currentThread().getName() + " puts " + cake);
        buffer[tail] = cake;
        tail = (tail + 1) % buffer.length;
        count++;
        notifyAll();
    }

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

        String cake = buffer[head];
        head = (head + 1) % buffer.length;
        count--;
        notifyAll();
        System.out.println(Thread.currentThread().getName() + " takes " + cake);
        return cake;
    }
}

运行结果

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
MakerThread-3 puts [ Cake No.20 by MakerThread-3 ]
EaterThread-3 takes [ Cake No.20 by MakerThread-3 ]
MakerThread-2 puts [ Cake No.21 by MakerThread-2 ]
MakerThread-1 puts [ Cake No.22 by MakerThread-1 ]
EaterThread-3 takes [ Cake No.21 by MakerThread-2 ]
EaterThread-1 takes [ Cake No.22 by MakerThread-1 ]
MakerThread-3 puts [ Cake No.23 by MakerThread-3 ]
EaterThread-2 takes [ Cake No.23 by MakerThread-3 ]
MakerThread-1 puts [ Cake No.24 by MakerThread-1 ]
MakerThread-2 puts [ Cake No.25 by MakerThread-2 ]
EaterThread-3 takes [ Cake No.24 by MakerThread-1 ]
EaterThread-3 takes [ Cake No.25 by MakerThread-2 ]
MakerThread-2 puts [ Cake No.26 by MakerThread-2 ]
MakerThread-1 puts [ Cake No.27 by MakerThread-1 ]
EaterThread-1 takes [ Cake No.26 by MakerThread-2 ]
EaterThread-2 takes [ Cake No.27 by MakerThread-1 ]
MakerThread-2 puts [ Cake No.28 by MakerThread-2 ]
EaterThread-2 takes [ Cake No.28 by MakerThread-2 ]
MakerThread-3 puts [ Cake No.29 by MakerThread-3 ]
EaterThread-3 takes [ Cake No.29 by MakerThread-3 ]
MakerThread-2 puts [ Cake No.30 by MakerThread-2 ]
EaterThread-1 takes [ Cake No.30 by MakerThread-2 ]
MakerThread-2 puts [ Cake No.31 by MakerThread-2 ]
MakerThread-1 puts [ Cake No.32 by MakerThread-1 ]
MakerThread-3 puts [ Cake No.33 by MakerThread-3 ]
EaterThread-1 takes [ Cake No.31 by MakerThread-2 ]
MakerThread-1 puts [ Cake No.34 by MakerThread-1 ]
EaterThread-2 takes [ Cake No.32 by MakerThread-1 ]
EaterThread-3 takes [ Cake No.33 by MakerThread-3 ]
MakerThread-1 puts [ Cake No.35 by MakerThread-1 ]
EaterThread-2 takes [ Cake No.34 by MakerThread-1 ]
EaterThread-1 takes [ Cake No.35 by MakerThread-1 ]
MakerThread-2 puts [ Cake No.36 by MakerThread-2 ]
EaterThread-1 takes [ Cake No.36 by MakerThread-2 ]

登场角色

Data

Data角色由Producer角色生成，供Consumer角色使用。在示例程序中，由String类扮演此角色。

Producer（生产者）

Producer角色生成Data角色，并将其传递给Channel角色。在示例程序中，由MakerThread扮演此角色。

Consumer（消费者）

Consumer角色从Channel角色获取Data角色并使用。在示例程序中，由EaterThread扮演此角色。

Channel（通道）

Channel角色保管从Producer角色获取的Data角色，还会响应Consumer角色的请求，传递Data角色。为了确保安全性，Channel角色会对Producer角色和Consumer角色的访问执行互斥处理。
当Producer角色将Data角色传递给Channel角色时，如果Channel角色的状态不适合接收Data角色，那么Producer角色将一直等待，直至Channel角色的状态变为可以接收为止。
当Consumer角色从Channel角色获取Data角色时，如果Channel角色中没有可以传递的Data角色，那么Consumer角色将一直等待，直至Channel的状态变为可以传递Data角色为止。
在示例程序中，由Table类扮演此角色。

类图

Fu1WQS.md.png