【数据结构与算法】(9)基础数据结构 之 阻塞队列的单锁实现、双锁实现详细代码示例讲解

目录

    • 2.8 阻塞队列
      • 1) 单锁实现
      • 2) 双锁实现

在这里插入图片描述

2.8 阻塞队列

之前的队列在很多场景下都不能很好地工作，例如

  1. 大部分场景要求分离向队列放入（生产者）、从队列拿出（消费者）两个角色、它们得由不同的线程来担当，而之前的实现根本没有考虑线程安全问题
  2. 队列为空，那么在之前的实现里会返回 null，如果就是硬要拿到一个元素呢？只能不断循环尝试
  3. 队列为满，那么再之前的实现里会返回 false，如果就是硬要塞入一个元素呢？只能不断循环尝试

因此我们需要解决的问题有

  1. 用锁保证线程安全
  2. 用条件变量让等待非空线程与等待不满线程进入等待状态，而不是不断循环尝试，让 CPU 空转

有同学对线程安全还没有足够的认识，下面举一个反例，两个线程都要执行入队操作（几乎在同一时刻）

public class TestThreadUnsafe {private final String[] array = new String[10];private int tail = 0;public void offer(String e) {array[tail] = e;tail++;}@Overridepublic String toString() {return Arrays.toString(array);}public static void main(String[] args) {TestThreadUnsafe queue = new TestThreadUnsafe();new Thread(()-> queue.offer("e1"), "t1").start();new Thread(()-> queue.offer("e2"), "t2").start();}
}

执行的时间序列如下，假设初始状态 tail = 0，在执行过程中由于 CPU 在两个线程之间切换，造成了指令交错

线程1线程2说明
array[tail]=e1线程1 向 tail 位置加入 e1 这个元素，但还没来得及执行 tail++
array[tail]=e2线程2 向 tail 位置加入 e2 这个元素，覆盖掉了 e1
tail++tail 自增为1
tail++tail 自增为2
最后状态 tail 为 2，数组为 [e2, null, null …]

糟糕的是，由于指令交错的顺序不同，得到的结果不止以上一种，宏观上造成混乱的效果

1) 单锁实现

Java 中要防止代码段交错执行，需要使用锁，有两种选择

  • synchronized 代码块，属于关键字级别提供锁保护，功能少
  • ReentrantLock 类，功能丰富

以 ReentrantLock 为例

ReentrantLock lock = new ReentrantLock();public void offer(String e) {lock.lockInterruptibly();try {array[tail] = e;tail++;} finally {lock.unlock();}
}

只要两个线程执行上段代码时，锁对象是同一个，就能保证 try 块内的代码的执行不会出现指令交错现象，即执行顺序只可能是下面两种情况之一

线程1线程2说明
lock.lockInterruptibly()t1对锁对象上锁
array[tail]=e1
lock.lockInterruptibly()即使 CPU 切换到线程2，但由于t1已经对该对象上锁，因此线程2卡在这儿进不去
tail++切换回线程1 执行后续代码
lock.unlock()线程1 解锁
array[tail]=e2线程2 此时才能获得锁，执行它的代码
tail++
  • 另一种情况是线程2 先获得锁，线程1 被挡在外面
  • 要明白保护的本质，本例中是保护的是 tail 位置读写的安全

事情还没有完，上面的例子是队列还没有放满的情况，考虑下面的代码（这回锁同时保护了 tail 和 size 的读写安全）

ReentrantLock lock = new ReentrantLock();
int size = 0;public void offer(String e) {lock.lockInterruptibly();try {if(isFull()) {// 满了怎么办?}array[tail] = e;tail++;size++;} finally {lock.unlock();}
}private boolean isFull() {return size == array.length;
}

之前是返回 false 表示添加失败，前面分析过想达到这么一种效果：

  • 在队列满时，不是立刻返回，而是当前线程进入等待
  • 什么时候队列不满了，再唤醒这个等待的线程，从上次的代码处继续向下运行

ReentrantLock 可以配合条件变量来实现，代码进化为

ReentrantLock lock = new ReentrantLock();
Condition tailWaits = lock.newCondition(); // 条件变量
int size = 0;public void offer(String e) {lock.lockInterruptibly();try {while (isFull()) {tailWaits.await();	// 当队列满时, 当前线程进入 tailWaits 等待}array[tail] = e;tail++;size++;} finally {lock.unlock();}
}private boolean isFull() {return size == array.length;
}
  • 条件变量底层也是个队列，用来存储这些需要等待的线程，当队列满了，就会将 offer 线程加入条件队列，并暂时释放锁
  • 将来我们的队列如果不满了（由 poll 线程那边得知）可以调用 tailWaits.signal() 来唤醒 tailWaits 中首个等待的线程，被唤醒的线程会再次抢到锁，从上次 await 处继续向下运行

思考为何要用 while 而不是 if，设队列容量是 3

操作前offer(4)offer(5)poll()操作后
[1 2 3]队列满，进入tailWaits 等待[1 2 3]
[1 2 3]取走 1，队列不满，唤醒线程[2 3]
[2 3]抢先获得锁，发现不满，放入 5[2 3 5]
[2 3 5]从上次等待处直接向下执行[2 3 5 ?]

关键点：

  • 从 tailWaits 中唤醒的线程，会与新来的 offer 的线程争抢锁，谁能抢到是不一定的，如果后者先抢到，就会导致条件又发生变化
  • 这种情况称之为虚假唤醒，唤醒后应该重新检查条件，看是不是得重新进入等待

最后的实现代码

/*** 单锁实现* @param <E> 元素类型*/
public class BlockingQueue1<E> implements BlockingQueue<E> {private final E[] array;private int head = 0;private int tail = 0;private int size = 0; // 元素个数@SuppressWarnings("all")public BlockingQueue1(int capacity) {array = (E[]) new Object[capacity];}ReentrantLock lock = new ReentrantLock();Condition tailWaits = lock.newCondition();Condition headWaits = lock.newCondition();@Overridepublic void offer(E e) throws InterruptedException {lock.lockInterruptibly();try {while (isFull()) {tailWaits.await();}array[tail] = e;if (++tail == array.length) {tail = 0;}size++;headWaits.signal();} finally {lock.unlock();}}@Overridepublic void offer(E e, long timeout) throws InterruptedException {lock.lockInterruptibly();try {long t = TimeUnit.MILLISECONDS.toNanos(timeout);while (isFull()) {if (t <= 0) {return;}t = tailWaits.awaitNanos(t);}array[tail] = e;if (++tail == array.length) {tail = 0;}size++;headWaits.signal();} finally {lock.unlock();}}@Overridepublic E poll() throws InterruptedException {lock.lockInterruptibly();try {while (isEmpty()) {headWaits.await();}E e = array[head];array[head] = null; // help GCif (++head == array.length) {head = 0;}size--;tailWaits.signal();return e;} finally {lock.unlock();}}private boolean isEmpty() {return size == 0;}private boolean isFull() {return size == array.length;}
}
  • public void offer(E e, long timeout) throws InterruptedException 是带超时的版本，可以只等待一段时间，而不是永久等下去，类似的 poll 也可以做带超时的版本，这个留给大家了

注意

  • JDK 中 BlockingQueue 接口的方法命名与我的示例有些差异
    • 方法 offer(E e) 是非阻塞的实现，阻塞实现方法为 put(E e)
    • 方法 poll() 是非阻塞的实现，阻塞实现方法为 take()

2) 双锁实现

单锁的缺点在于：

  • 生产和消费几乎是不冲突的，唯一冲突的是生产者和消费者它们有可能同时修改 size
  • 冲突的主要是生产者之间：多个 offer 线程修改 tail
  • 冲突的还有消费者之间：多个 poll 线程修改 head

如果希望进一步提高性能，可以用两把锁

  • 一把锁保护 tail
  • 另一把锁保护 head
ReentrantLock headLock = new ReentrantLock();  // 保护 head 的锁
Condition headWaits = headLock.newCondition(); // 队列空时，需要等待的线程集合ReentrantLock tailLock = new ReentrantLock();  // 保护 tail 的锁
Condition tailWaits = tailLock.newCondition(); // 队列满时，需要等待的线程集合

先看看 offer 方法的初步实现

@Override
public void offer(E e) throws InterruptedException {tailLock.lockInterruptibly();try {// 队列满等待while (isFull()) {tailWaits.await();}// 不满则入队array[tail] = e;if (++tail == array.length) {tail = 0;}// 修改 size （有问题）size++;} finally {tailLock.unlock();}
}

上面代码的缺点是 size 并不受 tailLock 保护，tailLock 与 headLock 是两把不同的锁，并不能实现互斥的效果。因此，size 需要用下面的代码保证原子性

AtomicInteger size = new AtomicInteger(0);	   // 保护 size 的原子变量size.getAndIncrement(); // 自增
size.getAndDecrement(); // 自减

代码修改为

@Override
public void offer(E e) throws InterruptedException {tailLock.lockInterruptibly();try {// 队列满等待while (isFull()) {tailWaits.await();}// 不满则入队array[tail] = e;if (++tail == array.length) {tail = 0;}// 修改 sizesize.getAndIncrement();} finally {tailLock.unlock();}
}

对称地，可以写出 poll 方法

@Override
public E poll() throws InterruptedException {E e;headLock.lockInterruptibly();try {// 队列空等待while (isEmpty()) {headWaits.await();}// 不空则出队e = array[head];if (++head == array.length) {head = 0;}// 修改 sizesize.getAndDecrement();} finally {headLock.unlock();}return e;
}

下面来看一个难题，就是如何通知 headWaits 和 tailWaits 中等待的线程，比如 poll 方法拿走一个元素，通知 tailWaits：我拿走一个，不满了噢，你们可以放了，因此代码改为

@Override
public E poll() throws InterruptedException {E e;headLock.lockInterruptibly();try {// 队列空等待while (isEmpty()) {headWaits.await();}// 不空则出队e = array[head];if (++head == array.length) {head = 0;}// 修改 sizesize.getAndDecrement();// 通知 tailWaits 不满（有问题）tailWaits.signal();} finally {headLock.unlock();}return e;
}

问题在于要使用这些条件变量的 await()， signal() 等方法需要先获得与之关联的锁，上面的代码若直接运行会出现以下错误

java.lang.IllegalMonitorStateException

那有同学说，加上锁不就行了吗，于是写出了下面的代码

在这里插入图片描述

发现什么问题了？两把锁这么嵌套使用，非常容易出现死锁，如下所示

在这里插入图片描述

因此得避免嵌套，两段加锁的代码变成了下面平级的样子

在这里插入图片描述

性能还可以进一步提升

  1. 代码调整后 offer 并没有同时获取 tailLock 和 headLock 两把锁，因此两次加锁之间会有空隙，这个空隙内可能有其它的 offer 线程添加了更多的元素，那么这些线程都要执行 signal()，通知 poll 线程队列非空吗？

    • 每次调用 signal() 都需要这些 offer 线程先获得 headLock 锁，成本较高，要想法减少 offer 线程获得 headLock 锁的次数
    • 可以加一个条件：当 offer 增加前队列为空，即从 0 变化到不空，才由此 offer 线程来通知 headWaits，其它情况不归它管

  2. 队列从 0 变化到不空，会唤醒一个等待的 poll 线程，这个线程被唤醒后，肯定能拿到 headLock 锁，因此它具备了唤醒 headWaits 上其它 poll 线程的先决条件。如果检查出此时有其它 offer 线程新增了元素（不空，但不是从0变化而来），那么不妨由此 poll 线程来唤醒其它 poll 线程

这个技巧被称之为级联通知（cascading notifies），类似的原因

  1. 在 poll 时队列从满变化到不满，才由此 poll 线程来唤醒一个等待的 offer 线程，目的也是为了减少 poll 线程对 tailLock 上锁次数，剩下等待的 offer 线程由这个 offer 线程间接唤醒

最终的代码为

public class BlockingQueue2<E> implements BlockingQueue<E> {private final E[] array;private int head = 0;private int tail = 0;private final AtomicInteger size = new AtomicInteger(0);ReentrantLock headLock = new ReentrantLock();Condition headWaits = headLock.newCondition();ReentrantLock tailLock = new ReentrantLock();Condition tailWaits = tailLock.newCondition();public BlockingQueue2(int capacity) {this.array = (E[]) new Object[capacity];}@Overridepublic void offer(E e) throws InterruptedException {int c;tailLock.lockInterruptibly();try {while (isFull()) {tailWaits.await();}array[tail] = e;if (++tail == array.length) {tail = 0;}            c = size.getAndIncrement();// a. 队列不满, 但不是从满->不满, 由此offer线程唤醒其它offer线程if (c + 1 < array.length) {tailWaits.signal();}} finally {tailLock.unlock();}// b. 从0->不空, 由此offer线程唤醒等待的poll线程if (c == 0) {headLock.lock();try {headWaits.signal();} finally {headLock.unlock();}}}@Overridepublic E poll() throws InterruptedException {E e;int c;headLock.lockInterruptibly();try {while (isEmpty()) {headWaits.await(); }e = array[head]; if (++head == array.length) {head = 0;}c = size.getAndDecrement();// b. 队列不空, 但不是从0变化到不空，由此poll线程通知其它poll线程if (c > 1) {headWaits.signal();}} finally {headLock.unlock();}// a. 从满->不满, 由此poll线程唤醒等待的offer线程if (c == array.length) {tailLock.lock();try {tailWaits.signal();} finally {tailLock.unlock();}}return e;}private boolean isEmpty() {return size.get() == 0;}private boolean isFull() {return size.get() == array.length;}}

双锁实现的非常精巧，据说作者 Doug Lea 花了一年的时间才完善了此段代码

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/452666.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Day 2.几个简单的函数接口 今日份浅学

Day 2.几个简单的函数接口 今日份浅学

1.函数接口 &#xff08;1&#xff09;.fgetc&#xff1a; int fgetc(FILE *stream); 功能&#xff1a;从流中度区下一个字符 参数&#xff1a; stream&#xff1a;文件流指针 返回&#xff1a; 成功返回ASCII值 失败返回 EOF 读到文件末尾返回EOF 练习&#xff1a; 读出文…
阅读更多...
基于C/C++的 FindOneOf 查找函数的未知bug

基于C/C++的 FindOneOf 查找函数的未知bug

CString str "Martel USB to Serial(COM5)";str.FindOneOf("COM");
阅读更多...
leetcode （算法）66.加一（python版）

leetcode （算法）66.加一（python版）

需求 给定一个由 整数 组成的 非空 数组所表示的非负整数&#xff0c;在该数的基础上加一。 最高位数字存放在数组的首位&#xff0c; 数组中每个元素只存储单个数字。 你可以假设除了整数 0 之外&#xff0c;这个整数不会以零开头。 示例 1&#xff1a; 输入&#xff1a;digi…
阅读更多...
2024美赛数学建模C题思路+代码

2024美赛数学建模C题思路+代码

文章目录 1 赛题思路2 美赛比赛日期和时间3 赛题类型4 美赛常见数模问题5 建模资料 1 赛题思路 (赛题出来以后第一时间在CSDN分享) https://blog.csdn.net/dc_sinor?typeblog 2 美赛比赛日期和时间 比赛开始时间&#xff1a;北京时间2024年2月2日&#xff08;周五&#xff…
阅读更多...
红外模块详解

红外模块详解

和红外有关的模块有很多&#xff0c;比如红外循迹&#xff0c;红外感应&#xff0c;红外发射&#xff0c;红外接收&#xff0c;红外对射&#xff0c;红外编解码等等。 今天我们要介绍的是红外编解码模块&#xff0c;它最常见的应用就是我们家里的电视、空调&#xff0c;当我们…
阅读更多...
手工方式安装19.22RU

手工方式安装19.22RU

使用手工方式打RU19.22 参考文档&#xff1a; Supplemental Readme - Grid Infrastructure Release Update 12.2.0.1.x / 18c /19c (Doc ID 2246888.1) 操作步骤&#xff1a; 1 Stop the CRS managed resources running from DB homes. 2 Run the pre root script. 3 Patch G…
阅读更多...
不要在吉利银河E8、星纪元ES之间瞎选

不要在吉利银河E8、星纪元ES之间瞎选

文 | AUTO芯球 作者 | 李诞 吉利银河E8和星纪元ES这两款车要怎么选 这是什么问题&#xff1f; 你看着这价格 吉利银河E8 是17.58-22.88万元 星纪元es是19.88-33.98万元 你要用E8高配对比ES低配&#xff1f; 好&#xff01; 想买这两款车的朋友 看完我说的 再做决定也…
阅读更多...
Xcode 15 及以上版本：libarclite 库缺少问题

Xcode 15 及以上版本：libarclite 库缺少问题

参考链接&#xff1a;Xcode 15 libarclite 缺失问题_sdk does not contain libarclite at the path /ap-CSDN博客 报错: SDK does not contain libarclite at the path /Applications/Xcode.app/Contents/Developer/Toolchains/XcodeDefault.xctoolchain/usr/lib/arc/libarcl…
阅读更多...
基于YOLOv7算法的高精度实时课堂场景下人脸检测系统（PyTorch+Pyside6+YOLOv7）

基于YOLOv7算法的高精度实时课堂场景下人脸检测系统（PyTorch+Pyside6+YOLOv7）

摘要&#xff1a;基于YOLOv7算法的高精度实时课堂场景下人脸检测系统可用于日常生活中检测与定位人脸&#xff0c;此系统可完成对输入图片、视频、文件夹以及摄像头方式的目标检测与识别&#xff0c;同时本系统还支持检测结果可视化与导出。本系统采用YOLOv7目标检测算法来训练…
阅读更多...
QSlider使用笔记

QSlider使用笔记

最近做项目使用到QSlider滑动条控件&#xff0c;在使用过的过程中&#xff0c;发现一个问题就是点滑动条上的一个位置&#xff0c;滑块并没有移动到鼠标点击的位置&#xff0c;体验感很差&#xff0c;于是研究了下&#xff0c;让鼠标点击后滑块移动到鼠标点击的位置。 1、event…
阅读更多...
ref和reactive, toRefs的使用

ref和reactive, toRefs的使用

看尤雨溪说&#xff1a;为什么Vue3 中应该使用 Ref 而不是 Reactive&#xff1f; toRefs import { ref, toRefs } from vue;// 定义一个响应式对象 const state ref({count: 0,name: Vue });// 使用toRefs转换为响应式引用对象 const reactiveState toRefs(state);// 现在你…
阅读更多...
二分查找------蓝桥杯

二分查找------蓝桥杯

题目描述&#xff1a; 请实现无重复数字的升序数组的二分查找 给定一个元素升序的、无重复数字的整型数组 nums 和一个目标值 target&#xff0c;写一个函数搜索 nums 中的target&#xff0c;如果目标值存在返回下标 (下标从0 开始)&#xff0c;否则返回-1 数据范围: 0 < l…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023