【Java|多线程与高并发】阻塞队列以及生产者-消费者模型

不念不忘少年蓝@ 2023-10-12 20:02 32阅读 0赞

#### 文章目录 ####

*  1. 前言
 *  2. 阻塞队列
 *  3. 实现阻塞队列
 *  4. 生产者-消费者模型
 *  5. 总结

## 1. 前言 ##

阻塞队列(BlockingQueue)常用于多线程编程中，可以实现线程之间的同步和协作。它可以用来解决生产者-消费者问题，其中生产者线程将元素插入队列，消费者线程从队列中获取元素，它们之间通过阻塞队列进行协调。  
![在这里插入图片描述][01d3d277a6674aaab309230f4022c4ac.gif_pic_center]

## 2. 阻塞队列 ##

Java中的阻塞队列(`BlockingQueue`)是一种特殊的队列，它在队列为空时会阻塞获取元素的操作，直到队列中有新的元素被添加进来；在队列已满时会阻塞插入元素的操作，直到队列中有空的位置。

需要注意的是 在Java中 `BlockingQueue`是一个接口  
![在这里插入图片描述][a9d280464fb5411fbae22050c5068043.png_pic_center]

Java中提供了多种阻塞队列的实现，包括：

1.  ArrayBlockingQueue：基于数组实现的有界阻塞队列，它按照先进先出的顺序对元素进行排序。
2.  LinkedBlockingQueue：基于链表实现的可选有界阻塞队列，它可以指定容量，如果不指定则默认为无界队列。
3.  PriorityBlockingQueue：基于优先级堆实现的无界阻塞队列，它可以按照元素的优先级进行排序。
4.  SynchronousQueue：一个不存储元素的阻塞队列，每个插入操作必须等待另一个线程的移除操作。
5.  LinkedBlockingDeque: 基于链表的双端阻塞队列。
6.  LinkedTransferQueue: 基于链表、无界的的阻塞队列。

作为一个队列,有三个基本操作,`入队,出队和查看队首元素.`

在使用阻塞队列时,有两点需要注意:

1.  offer和put可以实现入队列,offer并没有阻塞功能,put具有阻塞功能
2.  poll和take可以实现出队列,poll没有阻塞功能,take具有阻塞功能

示例:

public class Demo14 {
        
        public static void main(String[] args) throws InterruptedException {
        
            BlockingQueue<Integer> blockingQueue  = new LinkedBlockingQueue<>(10);
    
            blockingQueue.put(1);
            blockingQueue.put(2);
            int ret = blockingQueue.take();
            System.out.println(ret);
            ret = blockingQueue.take();
            System.out.println(ret);
            blockingQueue.take();
            System.out.println(ret);
        }
    }

代码分析:  
![在这里插入图片描述][72545de55ffe49d29737c6fdbf22edcb.png_pic_center]

## 3. 实现阻塞队列 ##

阻塞队列的实现并不复杂,主要是通过这个过程,强化对于阻塞队列的认识.

对于实现阻塞队列可以大致分为3步:

1.  实现一个普通的队列
2.  保证线程安全
3.  加上阻塞功能

实现一个普通队列的方法可以使用链表,也能够使用数组.

接下来就基于数组来实现一个 `循环队列`

代码:

public class MyBlockingQueue {
        
            private int[] elem;
            private int usedSize = 0;// 有效个数
            private int front = 0;
            private int rear = 0;
            public MyBlockingQueue(){
        
                this.elem = new int[10];
            }
            public MyBlockingQueue(int k){
        
                this.elem = new int[k];
            }
            /**
             * 入队
             * @param val
             * @return
             */
            public void put(int val){
        
                // 判断队列是否满了
                if (usedSize >= elem.length){
        
                    return;
                }
                // 进行入队列操作
                elem[rear] = val;
                rear++;
                if (rear >= elem.length){
        
                    rear = 0;
                }
                usedSize++;
            }
            /**
             * 出队
             * @return
             */
            public Integer take(){
        
                if (usedSize == 0){
        
                    return null;
                }
                int ret = elem[front];
                front++;
                if (front >= elem.length){
        
                    front = 0;
                }
                usedSize--;
                return ret;
            }
    
    }

循环队列在判断队列是否满时有两种方式:

1.  用计数器记录有效数据的个数(上述代码使用的是这种方法)
2.  浪费一个空间不用. 当`(rear+1) % 数组长度== front`时为队列满这种情况

以上就是一个简单的循环队列的实现,详细可以参考我的这篇文章[【数据结构与算法】队列-模拟实现队列以及设计循环队列][-]

解决了第一步,接下来就来实现第二步,**解决线程安全问题**

解决这个问题,就离不开`synchronized`了,以防万一再给变量加上`volatile`关键字

public class MyBlockingQueue {
        
            private int[] elem;
            private volatile int usedSize = 0;// 有效个数
            private volatile int front = 0;
            private volatile int rear = 0;
            public MyBlockingQueue(){
        
                this.elem = new int[10];
            }
            public MyBlockingQueue(int k){
        
                this.elem = new int[k];
            }
            /**
             * 入队
             * @param val
             * @return
             */
            public void put(int val){
        
                synchronized (this) {
        
                    // 判断队列是否满了
                    if (usedSize >= elem.length){
        
                        return;
                    }
                    // 进行入队列操作
                    elem[rear] = val;
                    rear++;
                    if (rear >= elem.length){
        
                        rear = 0;
                    }
                    usedSize++;
                }
            }
            /**
             * 出队
             * @return
             */
            public Integer take(){
        
                synchronized (this) {
        
                    if (usedSize == 0){
        
                        return null;
                    }
                    int ret = elem[front];
                    front++;
                    if (front >= elem.length){
        
                        front = 0;
                    }
                    usedSize--;
                    return ret;
                }
            }
    }

解决完线程安全问题,接下来就是`给put和take添加阻塞功能`.

以下两种场景会有阻塞功能:

1.  如果队列为空,出队列需要阻塞
2.  如果队列为满,入队列需要阻塞

阻塞功能很好加,直接使用`wait`方法即可

但除了阻塞,还要通知唤醒线程,. 例如线程为空,出队列在阻塞状态,而入队列之后,队列不为空,就要让出队列继续执行才行. 因此还要搭配`notify`来实现

阻塞队列实现代码如下:

public class MyBlockingQueue {
        
            private int[] elem;
            private volatile int usedSize = 0;// 有效个数
            private volatile int front = 0;
            private volatile int rear = 0;
            public MyBlockingQueue(){
        
                this.elem = new int[10];
            }
            public MyBlockingQueue(int k){
        
                this.elem = new int[k];
            }
            /**
             * 入队
             * @param val
             * @return
             */
            public void put(int val) throws InterruptedException {
        
                synchronized (this) {
        
                    // 判断队列是否满了
                    while (usedSize >= elem.length){
        
                        this.wait();
                    }
                    // 进行入队列操作
                    elem[rear] = val;
                    rear++;
                    if (rear >= elem.length){
        
                        rear = 0;
                    }
                    usedSize++;
                    this.notify();
                }
            }
            /**
             * 出队
             * @return
             */
            public Integer take() throws InterruptedException {
        
                synchronized (this) {
        
                    while (usedSize == 0){
        
                        this.wait();
                    }
                    int ret = elem[front];
                    front++;
                    if (front >= elem.length){
        
                        front = 0;
                    }
                    usedSize--;
                    this.notify();
                    return ret;
                }
            }
    }

代码分析:

![在这里插入图片描述][6571b5f6b5b044edad418f6e9d9b232b.png_pic_center]

此时出队列和入队列是相互唤醒的状态, 两个wait的执行条件互斥,因此不会出现同时阻塞的状态.

同时将wait的执行条件改为while,是因为如果线程是被`interrupt`唤醒的话,队列仍然为空,就不能去执行后续代码,因此要再进行条件判断,因此改为while更加稳妥

验证:  
![在这里插入图片描述][5142af5f546c4507af0fa87fa1abfb16.png_pic_center]  
使用阻塞队列的好处:

1.  使用阻塞队列,有利于代码"解耦合"
2.  削峰填谷,利用生产者消费者模型在并发量高的时候将这些并发量分配到每一个服务器上.

## 4. 生产者-消费者模型 ##

接下来来介绍生产者-消费者模型

> 生产者消费者模型是一种常见的并发编程模型，用于解决生产者和消费者之间的协作问题。在该模型中，生产者负责生产数据，消费者负责消费数据，它们通过共享的缓冲区来进行通信。

生产者消费者模型的主要特点：

1.  生产者和消费者是两个独立的实体，它们可以运行在不同的线程中。
2.  生产者负责生成数据，并将数据放入缓冲区中。
3.  消费者负责从缓冲区中取出数据，并进行相应的处理。
4.  缓冲区是生产者和消费者之间的共享数据结构，用于存储生产者生产的数据。
5.  生产者和消费者之间的操作是互斥的，即同一时间只能有一个生产者或一个消费者对缓冲区进行操作。

为了实现生产者消费者模型，可以使用阻塞队列来作为缓冲区。为了实现生产者消费者模型，可以**使用阻塞队列来作为缓冲区**。

生产者消费者模型的基本流程：

1.  创建一个共享的阻塞队列，作为生产者和消费者之间的缓冲区。
2.  创建生产者线程，它生成数据并将其放入队列中。
3.  创建消费者线程，它从队列中获取数据并进行处理。
4.  启动生产者和消费者线程，它们会并发执行。
5.  生产者线程生成数据并插入队列，当队列已满时会被阻塞。
6.  消费者线程从队列中获取数据并进行处理，当队列为空时会被阻塞。
7.  生产者和消费者线程可以通过阻塞队列进行同步和协作，生产者在队列已满时会等待，直到有空闲位置可以插入数据；消费者在队列为空时会等待，直到有数据可以获取。

代码示例:

public class Demo16 {
        
        public static void main(String[] args) {
        
            BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();
            // 生产者
            Thread t1 = new Thread(() ->{
        
                for (int i = 0; i < 1000; i++) {
        
                    try {
        
                        blockingQueue.put(i);
                        System.out.println("生产元素: "+ i);
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
        
                        throw new RuntimeException(e);
                    }
                }
            });
            t1.start();
    
            // 消费者
            Thread t2 = new Thread(() ->{
        
                while(true){
        
                    try {
        
                        Integer ret = blockingQueue.take();
                        System.out.println("消费元素: "+ ret);
                    } catch (InterruptedException e) {
        
                        throw new RuntimeException(e);
                    }
                }
            });
            t2.start();
        }
    }

运行截图:

![在这里插入图片描述][8dfa121585144f79b7adb77a25adb787.png_pic_center]  
上述代码仅仅只是一个示例,并没有什么实际意义.

生产者消费者模型依旧很重要,它可以有效地实现线程间的协作和资源共享，提高系统的并发性和吞吐量。

## 5. 总结 ##

阻塞队列的使用可以简化多线程编程的复杂性，避免手动实现线程间的同步和协作逻辑，提高代码的可读性和可维护性。基于阻塞队列的生产者-消费者模型也要重点掌握.。阻塞队列作为生产者和消费者之间的缓冲区，提供线程安全的插入和获取操作，并在队列为空或队列已满时进行阻塞，从而实现线程间的同步。  
![在这里插入图片描述][fb260d68d6a4443b928563ce9320c99a.gif_pic_center]

> 感谢你的观看!希望这篇文章能帮到你!  
> 专栏: [《从零开始的Java学习之旅》][Java]在不断更新中,欢迎订阅!  
> “愿与君共勉,携手共进!”  
> ![在这里插入图片描述][8fbf2a7f2d0e4db782e58035677a303d.png_pic_center]

[01d3d277a6674aaab309230f4022c4ac.gif_pic_center]: https://img-blog.csdnimg.cn/01d3d277a6674aaab309230f4022c4ac.gif#pic_center
[a9d280464fb5411fbae22050c5068043.png_pic_center]: https://img-blog.csdnimg.cn/a9d280464fb5411fbae22050c5068043.png#pic_center
[72545de55ffe49d29737c6fdbf22edcb.png_pic_center]: https://img-blog.csdnimg.cn/72545de55ffe49d29737c6fdbf22edcb.png#pic_center
[-]: https://blog.csdn.net/m0_63463510/article/details/128782522
[6571b5f6b5b044edad418f6e9d9b232b.png_pic_center]: https://img-blog.csdnimg.cn/6571b5f6b5b044edad418f6e9d9b232b.png#pic_center
[5142af5f546c4507af0fa87fa1abfb16.png_pic_center]: https://img-blog.csdnimg.cn/5142af5f546c4507af0fa87fa1abfb16.png#pic_center
[8dfa121585144f79b7adb77a25adb787.png_pic_center]: https://img-blog.csdnimg.cn/8dfa121585144f79b7adb77a25adb787.png#pic_center
[fb260d68d6a4443b928563ce9320c99a.gif_pic_center]: https://img-blog.csdnimg.cn/fb260d68d6a4443b928563ce9320c99a.gif#pic_center
[Java]: https://blog.csdn.net/m0_63463510/category_11797418.html
[8fbf2a7f2d0e4db782e58035677a303d.png_pic_center]: https://img-blog.csdnimg.cn/8fbf2a7f2d0e4db782e58035677a303d.png#pic_center