手写生产者消费者模型

# 生产者消费者模式说明：

生产者只在仓库未满时进行生产，仓库满时生产者进程被阻塞；消费者只在仓库非空时进行消费，仓库为空时消费者进程被阻塞；

# 实现的关键：

共享内存中的两个同步方法，及同步方法中wait()方法的调用。

synchronized 保证了对象只能被一个线程占用。 wait 保证了当线程在等待过程中释放锁，使得其他对象有机会获得锁。

# 代码实现：

代码包含这几个部分：仓库、产品、生产者、消费者、测试类

# 仓库代码

仓库代码核心方法

public synchronized void produce(T item):生产商品
public synchronized T consume():消费商品

/**
 * 仓库类，用于管理产品的生产、消费和存储。
 */
public class Storage<T> {
    private int index = 0;
    private static final int MAX = 10;//最大容量
    private List<T> storages = new ArrayList<T>(MAX);//存储集合

    public synchronized void produce(T item) {
        while (index >= MAX) {// 判断仓库满了，则等待。
            try {
                System.out.println("仓库满了，等待中...");
                this.wait();
                System.out.println("仓库不满了，开始生产");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("生产>>" + item.toString());
        storages.add(item);
        index++;   //先添加item，在进行加1操作
        notify();  //生产完 唤醒在此对象监视器上等待的单个线程，即消费者线程
    }

    public synchronized T consume() {
        while (index <= 0) {// 判断仓库空了，则等待。
            try {
                System.out.println("仓库为空，等待中...");
                this.wait();
                System.out.println("仓库不为空，开始消费");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        index--;//先进行减1操作，再remove
        T item = storages.remove(index);
        System.out.println("消费>>" + item.toString());
        notify();
        return item;
    }
}

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 Phone {

    private int id;// 手机编号

    public Phone(int id) {
        this.id = id;
    }

    @Override
    public String toString() {
        return "手机编号：" + id;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13

# 生产者

public class Producer implements Runnable {

    private Storage<Phone> storage;
    
    public Producer(Storage<Phone> storage) {
        this.storage = storage;
    }

    public void run() {
        for(int i = 0;i<20;i++){
            storage.produce(new Phone(i));
            
            try {
                Thread.sleep(10);//每隔10毫秒生产一个产品
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 消费者

public class Consumer implements Runnable {

    private Storage<Phone> storage;
    
    public Consumer(Storage<Phone> storage) {
        this.storage = storage;
    }

    public void run() {
        for(int i = 0;i<20;i++){
            storage.consume();
            try {
                Thread.sleep(100);//每隔100毫秒消费一个
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 测试类

public class ProducerAndConsumer {

    public static void main(String[] args) {
        Storage<Phone> storage = new Storage<Phone>();
        
        new Thread(new Producer(storage)).start();
        new Thread(new Consumer(storage)).start();
    }
}

1
2
3
4
5
6
7
8
9

# 测试类

生产>>手机编号：0
消费>>手机编号：0
生产>>手机编号：1
生产>>手机编号：2
生产>>手机编号：3
生产>>手机编号：4
生产>>手机编号：5
生产>>手机编号：6
生产>>手机编号：7
生产>>手机编号：8
生产>>手机编号：9
消费>>手机编号：9
生产>>手机编号：10
生产>>手机编号：11
仓库满了，等待中...
消费>>手机编号：11
仓库不满了，开始生产
生产>>手机编号：12
仓库满了，等待中...
消费>>手机编号：12
仓库不满了，开始生产
生产>>手机编号：13
仓库满了，等待中...
消费>>手机编号：13
仓库不满了，开始生产
生产>>手机编号：14
仓库满了，等待中...
消费>>手机编号：14
仓库不满了，开始生产
生产>>手机编号：15
仓库满了，等待中...
消费>>手机编号：15
仓库不满了，开始生产
生产>>手机编号：16
仓库满了，等待中...
消费>>手机编号：16
仓库不满了，开始生产
生产>>手机编号：17
仓库满了，等待中...
消费>>手机编号：17
仓库不满了，开始生产
生产>>手机编号：18
仓库满了，等待中...
消费>>手机编号：18
仓库不满了，开始生产
生产>>手机编号：19
消费>>手机编号：19
消费>>手机编号：10
消费>>手机编号：8
消费>>手机编号：7
消费>>手机编号：6
消费>>手机编号：5
消费>>手机编号：4
消费>>手机编号：3
消费>>手机编号：2
消费>>手机编号：1

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

# 代码关键点分析

# 锁问题

仓库类 Storage 中方法produce(T item) 和consume()要使用关键字synchronized修饰。使得变量index在多线程环境下的修改时线程安全的。

public synchronized void produce(T item):生产商品
public synchronized T consume():消费商品

# 线程的等待与重启

生产者线程的等待： 在生产时达到仓库的最大值后，要停止生产，此时需要调用 wait()方法。

while (index >= MAX) {// 判断仓库满了，则等待。
            try {
                System.out.println("仓库满了，等待中...");
                this.wait();
                System.out.println("仓库不满了，开始生产");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

1
2
3
4
5
6
7
8
9

生产者线程的重启：消费后，调用notify()方法唤醒等待当前锁的线程。

        index--;//先进行减1操作，再remove
        T item = storages.remove(index);
        System.out.println("消费>>" + item.toString());
        notify();

1
2
3
4

# 循环放置在`Producer`和`Consumer`类的`run`方法中

模拟多次生产/消费。

for(int i = 0;i<20;i++){
            storage.consume();
            try {
                Thread.sleep(100);//每隔100毫秒消费一个
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

1
2
3
4
5
6
7
8

# 时间延迟问题

Thread.sleep(10);s设置每隔10毫秒生产一个产品，模拟生产和消费的随机过程，还要设置时间延迟，否则会发生只有生产达到最大值后才会开始消费。

public void run() {
        for(int i = 0;i<20;i++){
            storage.produce(new Phone(i));
            
            try {
                Thread.sleep(10);//每隔10毫秒生产一个产品
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12

编辑

上次更新: 2022/05/19, 21:26:01

← Java 线程有哪些常用方法？ ThreadLocal 实现原理是什么？为什么要使用弱引用？→

手写生产者消费者模型

# 生产者消费者模式说明：

# 实现的关键：

# 代码实现：

# 仓库代码

# 产品

# 生产者

# 消费者

# 测试类

# 测试类

# 代码关键点分析

# 锁问题

# 线程的等待与重启

# 循环放置在Producer和Consumer类的run方法中

# 时间延迟问题

# 循环放置在`Producer`和`Consumer`类的`run`方法中