Java多线程08——生产者/消费者模式

头发还在

发布于 2023-10-16 10:59:13

1550

发布于 2023-10-16 10:59:13

文章被收录于专栏：桃花源桃花源

1 概念介绍

多线程环境下，我们经常需要多个线程的并发和相互通信。其中，有一个重要的多线程并发协作模型，即“生产者/消费者模式”。

2 角色介绍

生产者

负责生产数据的模块，可以是方法、对象、线程或进程。

消费者

负责处理数据的模块，可以是方法、对象、线程或进程。

缓冲区

消费者不能直接使用生产者的数据，在生产者和消费者之间有一个“缓冲区”。生产者将生产好的数据和内容放入“缓冲区”，消费者从“缓冲区”中取走要处理的数据。

缓冲区是实现并发的核心，设置缓冲区的优点：

实现线程的并发协作

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

解决闲忙不均，提高效率

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

3 实现生产者与消费者模式

3.1 创建要生产和消费的对象

public class Food {
    private int id;

    public Food(int id) {
        this.id = id;
    }
    
    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }
}

3.2 创建缓冲区

public class BufferBlock {
    //定义存放食物的盒子
    private Food[] foods = new Food[10];
    //定义操作盒子的索引
    private int index;
    /**
     * 放食物
     */
    public synchronized void push(Food food){
        //判断盒子是否已满
        while(this.index == this.foods.length){
            try {
                //释放锁
                this.wait();
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
        //唤醒领取食物的线程
        this.notify();
        this.foods[this.index] = food;
        this.index++;
    }
    /**
     * 取食物
     */
    public synchronized Food pop(){
        while(this.index == 0){
            try {
                //满足条件后，释放锁
                this.wait();
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
        //唤醒放入食物的线程
        this.notify();
        this.index--;
        return this.foods[this.index];
    }
}

3.3 创建生产者线程

public class ProducerThread extends Thread{
    private BufferBlock block;
    public ProducerThread(BufferBlock block){
        this.block = block;
    }
    @Override
    public void run() {
        for(int i=0;i<10;i++){
            Food food = new Food(i);
            this.block.push(food);
            System.out.println("生产食物：" + food.getId());
        }
    }
}

3.4 创建消费者线程

public class ConsumerThread extends Thread{
    private BufferBlock block;
    public ConsumerThread(BufferBlock block){
        this.block = block;
    }
    @Override
    public void run() {
        for(int i=0;i<10;i++){
            Food pop = this.block.pop();
            System.out.println("消费食物：" + pop.getId());
        }
    }
}

3.5 创建测试类

public class Test {
    public static void main(String[] args) {
        //创建缓冲区
        BufferBlock block = new BufferBlock();
        //启动消费者线程
        new ConsumerThread(block).start();
        //启动生产者线程
        new ProducerThread(block).start();
    }
}