Java JUC LinkedBlockingQueue解析

本文涉及的产品
全局流量管理 GTM,标准版 1个月
云解析 DNS,旗舰版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
简介: 阻塞队列 LinkedBlockingQueue

阻塞队列 LinkedBlockingQueue


介绍

上篇介绍了使用CAS算法实现的非阻塞队列ConcurrentLinkedQueue,本篇介绍的是使用独占锁实现的阻塞队列LinkedBlockingQueue。

1654827547349.png

该类图可以看到 LinkedBlockingQueue 也是使用单向链表实现的,其中包含head Node,last Node,用来存放头尾节点;并且还有一个初始值为 0 的原子变量count,用来记录队列元素个数;另外还包含两个 ReentrantLock 实例,分别用来控制元素入队和出队的原子性,其中 takeLock 用来控制同一时刻只有一个线程可以从队列头部获取元素,putLock 就是控制同一时刻只有一个线程可以从队列尾部添加元素。


notEmpty 和 notFull 是条件变量,他们内部都有一个条件队列来存放进队和出队时被阻塞的线程。


LinkedBlockingQueue 的构造函数如下:

public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);
}

可以看到默认情况下,LinkedBlockingQueue 队列容量为 int 最大值,当然我们也可以直接指定容量大小,所以在一定程度上可以说明 LinkedBlockingQueue 是有界阻塞队列


offer 操作

向队列尾部插入一个元素,如果队列有空闲则插入成功返回 true,如果队列已满则丢弃当前元素返回 false。如果插入的元素为 null 则抛出异常。并且该方法是非阻塞的。

public boolean offer(E e) {
    //(1)元素为空则抛出异常
    if (e == null) throw new NullPointerException();
     //(2)如果队列已满则丢弃元素
    final AtomicInteger count = this.count;
    if (count.get() == capacity)
        return false;
    int c = -1;
    //(3)构建新节点,然后获取putLock独占锁
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        //(4)如果队列不满则进入队列,并增加计数
        if (count.get() < capacity) {
            enqueue(node);
            c = count.getAndIncrement();
            //(5)
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        //(6)释放锁
        putLock.unlock();
    }
      //(7)
    if (c == 0)
        signalNotEmpty();
      //(8)
    return c >= 0;
}

代码(1)首先判断入队元素是否空,为空抛出异常。


代码(2)则判断是否队列已满,满的话就丢弃并返回 false。


代码(3)构建一个新的 Node 节点,然后获取 putLock 锁,当获取锁后其他线程调用 put 或 offer 方法都会被阻塞,放入 putLock 的 AQS 阻塞队列中。


代码(4)判断队列是否已满,为什么要再判断一次呢?因为在执行代码(2)到获取锁之间可能有其它线程通过 put 或 offer 添加了新元素,所以再判断一次是否队列已满,不满则新元素入队并且增加计数器。


代码(5)判断如果新元素入队后还有剩余空间,则唤醒 notFull 条件队列中正在阻塞的一个线程,(唤醒因为调用 notFull 的 await 操作的线程,比如执行了 put 方法而队列已满的时候)。


代码(6)释放获取的 putLock 锁,这里要注意,锁的释放一定要在 finally 里面做,因为即使 try 块抛出异常了,finally 也是会被执行到。另外释放锁后其他因为调用 put 操作而被阻塞的线程将会有一个获取到该锁。


代码(7)c == 0说明在执行代码(6)释放锁时队列至少有一个元素,队列里面有元素则执行 signalNotEmpty 方法。

private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

该方法的作用就是激活 notEmpty 的条件队列中因为调用 notEmpty 的 await 方法(比如调用 take 方法并且队列为空的时候)而被阻塞的一个线程,这也说明了调用条件变量的方法前要获取对应的锁。


总结:offer 方法通过 putLock 锁来保证添加的原子性,另外需要注意的就是在调用条件变量的时候需要先获取到对应的锁,并且入队只操作链表的尾结点。


put 操作

该方法向队列尾部插入一个元素,如果队列空闲则插入成功后直接返回,如果队列已满则阻塞当前线程,直到队列有空闲后插入成功返回。如果在阻塞时被其他线程设置了中断标志,则被阻塞线程会抛出异常然后返回。如果传入的元素是 null 则抛出异常

public void put(E e) throws InterruptedException {
    //(1)插入元素为空抛出异常
    if (e == null) throw new NullPointerException();
    int c = -1;
      //(2)构建新节点,并获取独占锁putLock
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        //(3)如果队列满则等待
        while (count.get() == capacity) {
            notFull.await();
        }
        //(4)插入队列并递增计数
        enqueue(node);
        c = count.getAndIncrement();
         //(5)
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        //(6)
        putLock.unlock();
    }
    //(7)
    if (c == 0)
        signalNotEmpty();
}

我们可以看到在代码(2)中使用putLock.lockInterruptibly()获取独占锁,相比在 offer 方法中获取独占锁的方法这个方法可以被中断。具体地说就是当前线程在获取锁的过程中,如果被其他线程设置了中断标志则当前线程会抛出 InterruptedException 异常,所以 put 操作在获取锁的过程中是可被中断的。


代码(3)判断当前队列已满,则调用notFull.await()方法把当前线程放入 notFull 的条件队列中,然后当前队列会释放 putLock 锁,由于 putLock 锁被释放了,所以别的线程就有机会获取到该锁。


使用 while 而不是 if 是为了防止虚假唤醒问题


poll 操作

从队列头部获取并移除一个元素,如果队列为空则返回 null,该方法是非阻塞方法。

public E poll() {
    final AtomicInteger count = this.count;
    //(1)队列为空则返回null
    if (count.get() == 0)
        return null;
    //(2)获取独占锁
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        //(3)队列不为空则出队并且递减计数
        if (count.get() > 0) {
            x = dequeue();
            c = count.getAndDecrement();
            //(4)
            if (c > 1)
                notEmpty.signal();
        }
    } finally {
        //(5)
        takeLock.unlock();
    }
    //(6)
    if (c == capacity)
        signalNotFull();
    //(7)
    return x;
}
private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

首先代码(1)很简单,先判断一下当前队列是否为空,为空直接返回 null。


代码(2)则获取独占锁 takeLock,当前线程获取该锁后,其他线程在调用 poll 或 take 方法时会被阻塞。


代码(3)判断如果当前队列不为空则进行出队操作,然后递减计数器。


虽然代码(3)中的判断队列是否为空和获取队列元素不是原子性的,但只有在 poll、take 或者 remove 操作的地方会递减 count 计数值,但是这三个方法都需要获取到 takeLock 锁才能进行操作,而当前线程已经获取了 takeLock 锁,所以其他线程没有机会在当前情况下递减 count 计数值,所以看起来不是原子性的,但是它们是线程安全的。


代码(4)判断如果c > 1则说明当前线程移除掉队列里面的一个元素后队列不为空(c 是删除元素前队列元素个数),那么这时候就可以激活因为调用 take 方法而被阻塞到 notEmpty 的条件队列里面的一个线程。


代码(6)说明当前线程移除队头元素前当前队列是满的,移除队头元素后当前队列至少有一个空闲位置,那么这时候就可以调用 signalNotFull 激活因为调用 put 方法而被阻塞到 notFull 的条件队列里的一个线程,signalNotFull 的代码如下。

private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

peek 操作

获取队列头部元素但是不从队列里面移除它,如果队列为空则返回 null。该方法是非阻塞方法。

public E peek() {
    //(1)
    if (count.get() == 0)
        return null;
     //(2)
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        Node<E> first = head.next;
        //(3)
        if (first == null)
            return null;
        else
            //(4)
            return first.item;
    } finally {
        takeLock.unlock();
    }
}

peek 方法不是很复杂,需要注意的就是这里需要再判断一下first == null,不能直接返回 first.item,因为代码(1)和代码(2)不是原子性的,很可能在代码(1)判断队列不为空后在获取锁前有其它线程执行了 poll 或 take 操作导致队列为空,然后直接返回 fist.item 就会空指针异常。


take 操作

该方法获取当前队列头部元素并从队列中移除它,如果队列为空则阻塞当前线程直到队列不为空然后获取并返回元素,如果在阻塞时被其他线程设置了中断标志,则阻塞线程会抛出异常。

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    //(1)获取锁
    takeLock.lockInterruptibly();
    try {
           //(2)队列为空则阻塞挂起
        while (count.get() == 0) {
            notEmpty.await();
        }
        //(3)出队并递减计数
        x = dequeue();
        c = count.getAndDecrement();
        //(4)
        if (c > 1)
            notEmpty.signal();
    } finally {
        //(5)
        takeLock.unlock();
    }
    //(6)
    if (c == capacity)
        signalNotFull();
    return x;
}

在代码(1)中,当前线程获取到独占锁,其他调用 take 或者 poll 操作的线程将会被阻塞挂起。


代码(2)判断如果队列为空则阻塞挂起当前线程,并把当前线程放入 notEmpty 的条件队列中。


代码(3)进行出队操作并递减计数。


代码(4)判断如果c > 1则说明当前队列不为空,那么唤醒 notEmpty 的条件队列里面的一个因为调用 take 操作而被阻塞的线程。


代码(5)释放锁。


代码(6)判断如果c == capacity则说明当前队列至少有一个空闲位置,那就激活 notFull 的条件队列里面的一个因为调用 put 操作而被阻塞的线程。


remove 操作

删除队列里面指定的元素,有则删除并返回 true,没有则返回 false。

public boolean remove(Object o) {
    if (o == null) return false;
    //(1)获取putLock 和 takeLock
    fullyLock();
    try {
        //(2)遍历寻找要删除的元素
        for (Node<E> trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
            //(3)
            if (o.equals(p.item)) {
                unlink(p, trail);
                return true;
            }
        }
        //(4)
        return false;
    } finally {
        //(5)
        fullyUnlock();
    }
}
void fullyLock() {
    putLock.lock();
    takeLock.lock();
}

首先代码(1)获取到双重锁,然后其他线程的入队和出队操作都会被挂起。


代码(2)遍历队列寻找要删除的元素,找不到返回 false,找到执行 unlink 方法,我们看一下这个方法。

void unlink(Node<E> p, Node<E> trail) {
        p.item = null;
        trail.next = p.next;
        if (last == p)
            last = trail;
        //如果当前队列满,再删除后也要唤醒等待的线程
        if (count.getAndDecrement() == capacity)
            notFull.signal();
}

trail 为删除元素的前驱节点,删除元素后,如果发现当前队列有空闲空间,则唤醒 notFull 的条件队列中的一个因为调用 put 方法而被阻塞的线程。


代码(5)调用 fullyUnlock 来进行释放双重锁。

void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

总结:由于 remove 方法在删除指定元素前加了两把锁,所以在遍历队列查找指定元素的过程中是线程安全的,并且此时其他调用入队、出队操作的线程全部会被阻塞。注意,获取多个资源锁的顺序与释放的顺序是相反的


size 操作

获取当前队列元素个数。

public int size() {
    return count.get();
}

由于出队和入队操作的 count 都是加锁的,所以结果相比 ConcurrentLinkedQueue 的 size 方法来说是准确的。


总结

image.png

LinkedBlockingQueue 的内部是通过单向链表实现的,使用头、尾节点来进行入队和出队操作,也就是入队操作都是对尾节点进行操作,出队操作都是对头节点进行操作。


对头、尾节点的操作分别使用了单独的独占锁从而保证了原子性,所以出队和入队操作是可以同时进行的。另外对头、尾节点的独占锁都配备了一个条件队列,用来存放被阻塞的线程,并结合入队、出队操作实现了一个生产消费模型。


相关文章
|
7天前
|
人工智能 自然语言处理 Java
FastExcel:开源的 JAVA 解析 Excel 工具,集成 AI 通过自然语言处理 Excel 文件,完全兼容 EasyExcel
FastExcel 是一款基于 Java 的高性能 Excel 处理工具,专注于优化大规模数据处理,提供简洁易用的 API 和流式操作能力,支持从 EasyExcel 无缝迁移。
55 9
FastExcel:开源的 JAVA 解析 Excel 工具,集成 AI 通过自然语言处理 Excel 文件,完全兼容 EasyExcel
|
14天前
|
存储 缓存 Java
Java 并发编程——volatile 关键字解析
本文介绍了Java线程中的`volatile`关键字及其与`synchronized`锁的区别。`volatile`保证了变量的可见性和一定的有序性,但不能保证原子性。它通过内存屏障实现,避免指令重排序,确保线程间数据一致。相比`synchronized`,`volatile`性能更优,适用于简单状态标记和某些特定场景,如单例模式中的双重检查锁定。文中还解释了Java内存模型的基本概念,包括主内存、工作内存及并发编程中的原子性、可见性和有序性。
Java 并发编程——volatile 关键字解析
|
12天前
|
Java 数据库连接 Spring
反射-----浅解析(Java)
在java中,我们可以通过反射机制,知道任何一个类的成员变量(成员属性)和成员方法,也可以堆任何一个对象,调用这个对象的任何属性和方法,更进一步我们还可以修改部分信息和。
|
1月前
|
存储 算法 Java
Java内存管理深度解析####
本文深入探讨了Java虚拟机(JVM)中的内存分配与垃圾回收机制,揭示了其高效管理内存的奥秘。文章首先概述了JVM内存模型,随后详细阐述了堆、栈、方法区等关键区域的作用及管理策略。在垃圾回收部分,重点介绍了标记-清除、复制算法、标记-整理等多种回收算法的工作原理及其适用场景,并通过实际案例分析了不同GC策略对应用性能的影响。对于开发者而言,理解这些原理有助于编写出更加高效、稳定的Java应用程序。 ####
|
运维 Java 大数据
Java并发JUC(java.util.concurrent)ForkJoin/异步回调
Java并发JUC(java.util.concurrent)ForkJoin/异步回调
Java并发JUC(java.util.concurrent)ForkJoin/异步回调
|
2天前
|
监控 Java
java异步判断线程池所有任务是否执行完
通过上述步骤,您可以在Java中实现异步判断线程池所有任务是否执行完毕。这种方法使用了 `CompletionService`来监控任务的完成情况,并通过一个独立线程异步检查所有任务的执行状态。这种设计不仅简洁高效,还能确保在大量任务处理时程序的稳定性和可维护性。希望本文能为您的开发工作提供实用的指导和帮助。
31 17
|
12天前
|
Java
Java—多线程实现生产消费者
本文介绍了多线程实现生产消费者模式的三个版本。Version1包含四个类:`Producer`(生产者)、`Consumer`(消费者)、`Resource`(公共资源)和`TestMain`(测试类)。通过`synchronized`和`wait/notify`机制控制线程同步,但存在多个生产者或消费者时可能出现多次生产和消费的问题。 Version2将`if`改为`while`,解决了多次生产和消费的问题,但仍可能因`notify()`随机唤醒线程而导致死锁。因此,引入了`notifyAll()`来唤醒所有等待线程,但这会带来性能问题。
Java—多线程实现生产消费者
|
14天前
|
安全 Java Kotlin
Java多线程——synchronized、volatile 保障可见性
Java多线程中,`synchronized` 和 `volatile` 关键字用于保障可见性。`synchronized` 保证原子性、可见性和有序性,通过锁机制确保线程安全;`volatile` 仅保证可见性和有序性,不保证原子性。代码示例展示了如何使用 `synchronized` 和 `volatile` 解决主线程无法感知子线程修改共享变量的问题。总结:`volatile` 确保不同线程对共享变量操作的可见性,使一个线程修改后,其他线程能立即看到最新值。
|
14天前
|
消息中间件 缓存 安全
Java多线程是什么
Java多线程简介:本文介绍了Java中常见的线程池类型,包括`newCachedThreadPool`(适用于短期异步任务)、`newFixedThreadPool`(适用于固定数量的长期任务)、`newScheduledThreadPool`(支持定时和周期性任务)以及`newSingleThreadExecutor`(保证任务顺序执行)。同时,文章还讲解了Java中的锁机制,如`synchronized`关键字、CAS操作及其实现方式,并详细描述了可重入锁`ReentrantLock`和读写锁`ReadWriteLock`的工作原理与应用场景。
|
14天前
|
安全 Java 编译器
深入理解Java中synchronized三种使用方式:助您写出线程安全的代码
`synchronized` 是 Java 中的关键字,用于实现线程同步,确保多个线程互斥访问共享资源。它通过内置的监视器锁机制,防止多个线程同时执行被 `synchronized` 修饰的方法或代码块。`synchronized` 可以修饰非静态方法、静态方法和代码块,分别锁定实例对象、类对象或指定的对象。其底层原理基于 JVM 的指令和对象的监视器,JDK 1.6 后引入了偏向锁、轻量级锁等优化措施,提高了性能。
40 3

热门文章

最新文章