Java Web 实战 08 - 多线程基础之阻塞队列

╰半橙微兮° 2024-03-25 10:14 73阅读 0赞

#### 阻塞队列 ####

*  阻塞队列
 *  1. 阻塞队列是什么 ?
 *  2. 生产者-消费者模型
 *  3. 标准库中的阻塞队列
 *  4. 自己实现一个阻塞队列

> 大家好 , 这篇文章给大家讲解的知识点是 : 阻塞队列 , 我们分成几个模块来了解阻塞队列
> 
> 1.  阻塞队列是什么 ?
> 2.  生产者 - 消费者模型是什么 ?
> 3.  标准库当中的阻塞队列是怎样实现的 ?
> 4.  自己怎样实现一个简单的阻塞队列 ?
> 
> 推荐大家跳转到[此链接][Link 1]进行观看  
> [上一篇文章的链接][Link 2]也给大家贴到这里了  
> ![在这里插入图片描述][76a293a20d3e4a7093491b27ac4c1446.png]

## 阻塞队列 ##

## 1. 阻塞队列是什么 ? ##

这里的阻塞队列和咱们内核中 , 表示阻塞状态的 PCB 的那个链表还不一样  
我们这里的阻塞队列 , 是能够保证线程安全的队列 , 内部采用了锁机制来保证线程安全  
他还有一个很大的特点 : 如果队列为空 , 尝试出队列 , 就会阻塞 ; 如果队列满 , 尝试入队列 , 也会阻塞

> 阻塞 : 就是让线程停下来等一等 , 本质上就是修改了线程的状态 , 让线程的 PCB 在内核中暂时不参与调度

其实就相当于餐厅营业 , 没人来吃饭了咱就关门停业 ? 餐厅人多 , 咱们可以让顾客稍微等一会

相对应的还有  
无锁队列 : 也是一种线程安全的队列 , 但是实现内部没有使用锁 , 相比阻塞队列来说更加高效 , 需要消耗更多的 CPU 资源  
消息队列 : 在队列中涵盖多种不同 “类型” 的元素 , 取元素的时候可以按照某个类型来取 , 做到针对该类型的先进先出 , 甚至说会把消息队列作为服务器 , 单独部署 (常见的有 : kafka , rocketmq , rabbitmq)

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

![image.png][]  
使用生产者消费者模型 , 在工作中是非常频繁的

生产者-消费者模型的优点有很多 , 其中最明显的优点有两条 :

1.  可以做到更好的 “解耦合”

![image.png][image.png 1]

1.  能够做到 “削峰填谷” , 提高整个系统的抗风险能力

![image.png][image.png 2]

那么我们刚才介绍了解耦合 , 那再帮大家回顾一下高内聚的意思  
写一个功能的时候 , 尽量让这个功能的代码能够集中放置 , 不要东一块西一块  
类似于 “归类”  
举个栗子 :  
我们要是没对自己电脑的文件进行分门别类的话 , 到时候想要找一个文件就会特别困难 .  
这就是高内聚的反例  
如果我们把所有的文件分门别类的存储 , 把代码放到一个文件夹 , 把软件放到另一个文件夹 , 这样就类似于 “高内聚”

## 3. 标准库中的阻塞队列 ##

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class Demo20 {
        
        public static void main(String[] args) {
        
            // Java 标准库中的阻塞队列
            // BlockingQueue 是父类,我们要实例化他的接口
            BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
    
            // 消费者线程
            Thread customer = new Thread(() -> {
        
                while(true) {
        
                    // BlockingQueue 虽然也支持 offer、poll 等普通队列的方法
                    // 但是仍然推荐大家使用 put 来入队列,使用 take 来出队列(可以做到阻塞)
                    try {
        
                        int val = queue.take();// Integer->int 自动拆箱
                        System.out.println("消费元素:" + val);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
            customer.start();
    
            // 生产者线程
            Thread producer = new Thread(() -> {
        
                int n = 0;
                while(true) {
        
                    try {
        
                        System.out.println("生产元素:" + n);
                        queue.put(n);
                        n++;
                        Thread.sleep(500);// 方便观察
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
            producer.start();
        }
    }

![image.png][image.png 3]

## 4. 自己实现一个阻塞队列 ##

我们目前已经完成了最基本的队列的实现  
接下来 , 我们就实现阻塞队列  
阻塞队列需要满足两点 :

1.  线程安全 : 通过加锁实现

![image.png][image.png 4]

// 我们自己模拟一个阻塞队列
    // 基于数组的方式实现阻塞队列
    // 模拟两个方法:put入队列/take出队列
    class MyBlockingQueue {
        
        // 假设阻塞队列最大是 1000 个元素
        private int[] items = new int[1000];
        // 队首
        private int head = 0;
        // 队尾
        private int tail = 0;
        // 队列的元素个数
        private int size = 0;
    
        // 入队列
        public void put(int value) {
        
            synchronized (this) {
        
                if(size == items.length) {
        
                    // 队列已满,无法插入
                    return;
                }
                items[tail++] = value;
                // 如果 tail 达到数组末尾,就需要从头开始
                if(tail == items.length) {
        
                    tail = 0;
                }
                size++;
            }
        }
    
        // 出队列
        public Integer take() {
        
            int ret = 0;
            synchronized (this) {
        
                if(size == 0) {
        
                    // 队列为空,无法出队列
                    return null;
                }
                ret = items[head++];
                if(size == items.length) {
        
                    head = 0;
                }
                size--;
            }
            return ret;
        }
    }
    
    public class Demo21 {
        
    }

1.  阻塞 : 通过 wait() 实现
    
    1.  队列为空要阻塞 : 当队列不空的时候 , 就唤醒
    2.  队列未满要阻塞 : 当队列不满的时候 , 就唤醒

// 我们自己模拟一个阻塞队列
    // 基于数组的方式实现阻塞队列
    // 模拟两个方法:put入队列/take出队列
    class MyBlockingQueue {
        
        // 假设阻塞队列最大是 1000 个元素
        private int[] items = new int[1000];
        // 队首
        private int head = 0;
        // 队尾
        private int tail = 0;
        // 队列的元素个数
        private int size = 0;
    
        // 入队列
        public void put(int value) throws InterruptedException {
        
            synchronized (this) {
        
                if(size == items.length) {
        
                    // 队列满了,要进行阻塞
                    this.wait();
                }
                items[tail++] = value;
                // 如果 tail 达到数组末尾,就需要从头开始
                if(tail == items.length) {
        
                    tail = 0;
                }
                size++;
                // 队列不空的时候(插入元素成功),就需要唤醒
                // notify 即使没人在等待,多调用几次 notify 也没啥副作用
                this.notify();
            }
        }
    
        // 出队列
        public Integer take() throws InterruptedException {
        
            int ret = 0;
            synchronized (this) {
        
                if(size == 0) {
        
                    // 队列为空,就阻塞等待
                    this.wait();// wait 方法需要在 synchronized 里面使用
                }
                ret = items[head++];
                if(size == items.length) {
        
                    head = 0;
                }
                size--;
                // 队列不满(取走一个元素),就需要唤醒
                this.notify();
            }
            return ret;
        }
    }
    
    public class Demo21 {
        
        public static void main(String[] args) throws InterruptedException {
        
            MyBlockingQueue myQueue = new MyBlockingQueue();
            myQueue.put(100);
            myQueue.take();
        }
    }

![image.png][image.png 5]

然后 , 我们还需要在每个变量前面加上 `volatile` 关键字 , 因为这段代码既有读 , 又有写 , 为了防止内存可见性 , 所以加上 `volatile`关键字

最后 , 我们再进行一个优化操作  
![image.png][image.png 6]

接下来 , 我们就可以去模拟实现生产者-消费者模型

// 我们自己模拟一个阻塞队列
    // 基于数组的方式实现阻塞队列
    // 模拟两个方法:put入队列/take出队列
    class MyBlockingQueue {
        
        // 假设阻塞队列最大是 1000 个元素
        volatile private int[] items = new int[1000];
        // 队首
        volatile private int head = 0;
        // 队尾
        volatile private int tail = 0;
        // 队列的元素个数
        volatile private int size = 0;
    
        // 入队列
        public void put(int value) throws InterruptedException {
        
            synchronized (this) {
        
                while(size == items.length) {
        
                    // 队列满了,要进行阻塞
                    this.wait();
                }
                items[tail++] = value;
                // 如果 tail 达到数组末尾,就需要从头开始
                if(tail == items.length) {
        
                    tail = 0;
                }
                size++;
                // 队列不空的时候(插入元素成功),就需要唤醒
                // notify 即使没人在等待,多调用几次 notify 也没啥副作用
                this.notify();
            }
        }
    
        // 出队列
        public Integer take() throws InterruptedException {
        
            int ret = 0;
            synchronized (this) {
        
                while(size == 0) {
        
                    // 队列为空,就阻塞等待
                    this.wait();// wait 方法需要在 synchronized 里面使用
                }
                ret = items[head++];
                if(size == items.length) {
        
                    head = 0;
                }
                size--;
                // 队列不满(取走一个元素),就需要唤醒
                this.notify();
            }
            return ret;
        }
    }
    
    public class Demo21 {
        
        public static void main(String[] args) throws InterruptedException {
        
            MyBlockingQueue queue = new MyBlockingQueue();
    
            // 消费者
            Thread customer = new Thread(() -> {
        
                while(true) {
        
                    int value = 0;
                    try {
        
                        value = queue.take();
                        System.out.println("消费:" + value);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
            customer.start();
    
            // 生产者
            Thread producer = new Thread(() -> {
        
                int value = 0;
                while(true) {
        
                    try {
        
                        queue.put(value);
                        System.out.println("生产" + value);
                        value++;
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
            producer.start();
        }
    }

到这里 , 阻塞队列的讲解就完毕了  
对你有帮助的话请一键三连~  
![在这里插入图片描述][8304eedac9354116898ceac9667b4eea.png]

[Link 1]: https://www.yuque.com/jialebihaitao/study/qzym2pw332lm6k7q?singleDoc#%20%E3%80%8A2.%20%E5%A4%9A%E7%BA%BF%E7%A8%8B%20%28%E5%9F%BA%E7%A1%80%29%E3%80%8B
[Link 2]: https://blog.csdn.net/m0_53117341/article/details/129458421
[76a293a20d3e4a7093491b27ac4c1446.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/94132507b7dd4bc49826d5536c06b19e.png
[image.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/5c7de69627e54f3d94e4b6c5e338ec53.png
[image.png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/ccdcc8e0d94048ad8d6043e00ef1dac1.png
[image.png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/e62694da5d684441b0b477758384d100.png
[image.png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/7cd0d533330a4824a80361d068f40028.png
[image.png 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/a015e275618746c39f3640c0b2507504.png
[image.png 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/7d786cdde718406d9330f582e2a369d4.png
[image.png 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/1aeb12b9351f4dff949fbe04f6e8737f.png
[8304eedac9354116898ceac9667b4eea.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/4ac674010fcb4b6abde7f2df4a091517.png