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

刺骨的言语ヽ痛彻心扉 2024-03-26 12:55 73阅读 0赞

![e526072d744c32a40b2c0b8dd70e1ab4.png][]

## 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();
        }
    }

### 3.6 结果输出 ###

> 生产食物：0
> 
> 生产食物：1
> 
> 生产食物：2
> 
> 生产食物：3
> 
> 生产食物：4
> 
> 生产食物：5
> 
> 生产食物：6
> 
> 消费食物：0
> 
> 消费食物：6
> 
> 消费食物：5
> 
> 消费食物：4
> 
> 消费食物：3
> 
> 消费食物：2
> 
> 消费食物：1
> 
> 生产食物：7
> 
> 生产食物：8
> 
> 生产食物：9
> 
> 消费食物：9
> 
> 消费食物：8
> 
> 消费食物：7

## 4 总结 ##

**生产者消费者模式：**

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

2 对于生产者，没有生产数据之前，消费者要进入等待状态；而生产了数据之后，即需要马上通知消费者进行消费。

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

4 也可以看出缓冲区对于该模式的重要性，工作当中，更多的是使用消息队（各种MQ，如 kafka, ActiveMQ, RockerMQ, RabbitMQ等）列来实现。

[e526072d744c32a40b2c0b8dd70e1ab4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/2259e2cd130b4578af7bcf487f0c8f70.png