java多线程并发之旅-17--双端队列之 LinkedBlockingDeque

2023-10-22 05:32
文章标签 java 并发 多线程 队列 17 双端 之旅 linkedblockingdeque

本文主要是介绍java多线程并发之旅-17--双端队列之 LinkedBlockingDeque，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

问题

  • LinkedBlockingDeque 是什么？

  • 优缺点？

  • 应用场景？

  • 源码实现？

  • 个人启发？

LinkedBlockingDeque

双向并发阻塞队列。

所谓双向是指可以从队列的头和尾同时操作，并发只是线程安全的实现，阻塞允许在入队出队不满足条件时挂起线程，这里说的队列是指支持FIFO/FILO实现的链表。

  1. 要想支持阻塞功能，队列的容量一定是固定的，否则无法在入队的时候挂起线程。也就是capacity是final类型的。

  2. 既然是双向链表，每一个结点就需要前后两个引用，这样才能将所有元素串联起来，支持双向遍历。也即需要prev/next两个引用。

  3. 双向链表需要头尾同时操作，所以需要first/last两个节点，当然可以参考LinkedList那样采用一个节点的双向来完成，那样实现起来就稍微麻烦点。

  4. 既然要支持阻塞功能，就需要锁和条件变量来挂起线程。这里使用一个锁两个条件变量来完成此功能。

优缺点

优点当然是功能足够强大，同时由于采用一个独占锁，因此实现起来也比较简单。所有对队列的操作都加锁就可以完成。同时独占锁也能够很好的支持双向阻塞的特性。

凡事有利必有弊。缺点就是由于独占锁，所以不能同时进行两个操作，这样性能上就大打折扣。从性能的角度讲LinkedBlockingDeque要比LinkedQueue要低很多，比CocurrentLinkedQueue就低更多了，这在高并发情况下就比较明显了。

前面分析足够多的Queue实现后，LinkedBlockingDeque的原理和实现就不值得一提了，无非是在独占锁下对一个链表的普通操作。

序列化

有趣的是此类支持序列化，但是Node并不支持序列化，因此fist/last就不能序列化，那么如何完成序列化/反序列化过程呢？

private void writeObject(java.io.ObjectOutputStream s)throws java.io.IOException {lock.lock();try {// Write out capacity and any hidden stuffs.defaultWriteObject();// Write out all elements in the proper order.for (Node<E> p = first; p != null; p = p.next)s.writeObject(p.item);// Use trailing null as sentinels.writeObject(null);} finally {lock.unlock();}
}private void readObject(java.io.ObjectInputStream s)throws java.io.IOException, ClassNotFoundException {s.defaultReadObject();count = 0;first = null;last = null;// Read in all elements and place in queuefor (;;) {E item = (E)s.readObject();if (item == null)break;add(item);}
}

描述的是LinkedBlockingDeque序列化/反序列化的过程。序列化时将真正的元素写入输出流，最后还写入了一个null。读取的时候将所有对象列表读出来，如果读取到一个null就表示结束。这就是为什么写入的时候写入一个null的原因，因为没有将count写入流，所以就靠null来表示结束，省一个整数空间。

源码

接口

/*** @since 1.6* @author  Doug Lea* @param <E> the type of elements held in this collection*/
public class LinkedBlockingDeque<E>extends AbstractQueue<E>implements BlockingDeque<E>, java.io.Serializable {

双向链表节点

/** Doubly-linked list node class */
static final class Node<E> {/*** The item, or null if this node has been removed.*/E item;/*** One of:* - the real predecessor Node* - this Node, meaning the predecessor is tail* - null, meaning there is no predecessor*/Node<E> prev;/*** One of:* - the real successor Node* - this Node, meaning the successor is head* - null, meaning there is no successor*/Node<E> next;Node(E x) {item = x;}
}

基础属性

/*** Pointer to first node.* Invariant: (first == null && last == null) ||*            (first.prev == null && first.item != null)*/
transient Node<E> first;
/*** Pointer to last node.* Invariant: (first == null && last == null) ||*            (last.next == null && last.item != null)*/
transient Node<E> last;
/** Number of items in the deque */
private transient int count;
/** Maximum number of items in the deque */
private final int capacity;
/** Main lock guarding all access */
final ReentrantLock lock = new ReentrantLock();
/** Condition for waiting takes */
private final Condition notEmpty = lock.newCondition();
/** Condition for waiting puts */
private final Condition notFull = lock.newCondition();

构造器

    /*** Creates a {@code LinkedBlockingDeque} with a capacity of* {@link Integer#MAX_VALUE}.*/public LinkedBlockingDeque() {this(Integer.MAX_VALUE);}/*** Creates a {@code LinkedBlockingDeque} with the given (fixed) capacity.** @param capacity the capacity of this deque* @throws IllegalArgumentException if {@code capacity} is less than 1*/public LinkedBlockingDeque(int capacity) {if (capacity <= 0) throw new IllegalArgumentException();this.capacity = capacity;}/*** Creates a {@code LinkedBlockingDeque} with a capacity of* {@link Integer#MAX_VALUE}, initially containing the elements of* the given collection, added in traversal order of the* collection's iterator.** @param c the collection of elements to initially contain* @throws NullPointerException if the specified collection or any*         of its elements are null*/public LinkedBlockingDeque(Collection<? extends E> c) {this(Integer.MAX_VALUE);final ReentrantLock lock = this.lock;lock.lock(); // Never contended, but necessary for visibilitytry {for (E e : c) {if (e == null)throw new NullPointerException();if (!linkLast(new Node<E>(e)))throw new IllegalStateException("Deque full");}} finally {lock.unlock();}}

吐槽

默认竟然构造成最大整数，真是令人费解？

初始化线程安全保证

使用了 ReentrantLock 可互斥锁，来保证线程安全性。

看的出来，如果有元素为空，会直接抛出异常。

添加元素

    /*** @throws IllegalStateException if this deque is full* @throws NullPointerException {@inheritDoc}*/public void addFirst(E e) {if (!offerFirst(e))throw new IllegalStateException("Deque full");}/*** @throws IllegalStateException if this deque is full* @throws NullPointerException  {@inheritDoc}*/public void addLast(E e) {if (!offerLast(e))throw new IllegalStateException("Deque full");}/*** @throws NullPointerException {@inheritDoc}*/public boolean offerFirst(E e) {if (e == null) throw new NullPointerException();Node<E> node = new Node<E>(e);final ReentrantLock lock = this.lock;lock.lock();try {return linkFirst(node);} finally {lock.unlock();}}/*** @throws NullPointerException {@inheritDoc}*/public boolean offerLast(E e) {if (e == null) throw new NullPointerException();Node<E> node = new Node<E>(e);final ReentrantLock lock = this.lock;lock.lock();try {return linkLast(node);} finally {lock.unlock();}}/*** @throws NullPointerException {@inheritDoc}* @throws InterruptedException {@inheritDoc}*/public void putFirst(E e) throws InterruptedException {if (e == null) throw new NullPointerException();Node<E> node = new Node<E>(e);final ReentrantLock lock = this.lock;lock.lock();try {while (!linkFirst(node))notFull.await();} finally {lock.unlock();}}/*** @throws NullPointerException {@inheritDoc}* @throws InterruptedException {@inheritDoc}*/public void putLast(E e) throws InterruptedException {if (e == null) throw new NullPointerException();Node<E> node = new Node<E>(e);final ReentrantLock lock = this.lock;lock.lock();try {while (!linkLast(node))notFull.await();} finally {lock.unlock();}}/*** @throws NullPointerException {@inheritDoc}* @throws InterruptedException {@inheritDoc}*/public boolean offerFirst(E e, long timeout, TimeUnit unit)throws InterruptedException {if (e == null) throw new NullPointerException();Node<E> node = new Node<E>(e);long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (!linkFirst(node)) {if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}return true;} finally {lock.unlock();}}/*** @throws NullPointerException {@inheritDoc}* @throws InterruptedException {@inheritDoc}*/public boolean offerLast(E e, long timeout, TimeUnit unit)throws InterruptedException {if (e == null) throw new NullPointerException();Node<E> node = new Node<E>(e);long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (!linkLast(node)) {if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}return true;} finally {lock.unlock();}}

linkFirst & linkLast

这里使用了 Condition 类来保证队列阻塞。

见 阻塞队列实现原理

    /*** Links node as first element, or returns false if full.*/private boolean linkFirst(Node<E> node) {// assert lock.isHeldByCurrentThread();if (count >= capacity)return false;Node<E> f = first;node.next = f;first = node;if (last == null)last = node;elsef.prev = node;++count;notEmpty.signal();return true;}/*** Links node as last element, or returns false if full.*/private boolean linkLast(Node<E> node) {// assert lock.isHeldByCurrentThread();if (count >= capacity)return false;Node<E> l = last;node.prev = l;last = node;if (first == null)first = node;elsel.next = node;++count;notEmpty.signal();return true;}

移除元素

    /*** @throws NoSuchElementException {@inheritDoc}*/public E removeFirst() {E x = pollFirst();if (x == null) throw new NoSuchElementException();return x;}/*** @throws NoSuchElementException {@inheritDoc}*/public E removeLast() {E x = pollLast();if (x == null) throw new NoSuchElementException();return x;}public E pollFirst() {final ReentrantLock lock = this.lock;lock.lock();try {return unlinkFirst();} finally {lock.unlock();}}public E pollLast() {final ReentrantLock lock = this.lock;lock.lock();try {return unlinkLast();} finally {lock.unlock();}}

unlinkLast & unlinkFirst

原理和 linkFirst 是类似的，仍然使用 Condition 保证阻塞。

    /*** Removes and returns first element, or null if empty.*/private E unlinkFirst() {// assert lock.isHeldByCurrentThread();Node<E> f = first;if (f == null)return null;Node<E> n = f.next;E item = f.item;f.item = null;f.next = f; // help GCfirst = n;if (n == null)last = null;elsen.prev = null;--count;notFull.signal();return item;}/*** Removes and returns last element, or null if empty.*/private E unlinkLast() {// assert lock.isHeldByCurrentThread();Node<E> l = last;if (l == null)return null;Node<E> p = l.prev;E item = l.item;l.item = null;l.prev = l; // help GClast = p;if (p == null)first = null;elsep.next = null;--count;notFull.signal();return item;}/*** Unlinks x.*/void unlink(Node<E> x) {// assert lock.isHeldByCurrentThread();Node<E> p = x.prev;Node<E> n = x.next;if (p == null) {unlinkFirst();} else if (n == null) {unlinkLast();} else {p.next = n;n.prev = p;x.item = null;// Don't mess with x's links.  They may still be in use by// an iterator.--count;notFull.signal();}}

获取元素

    public E takeFirst() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lock();try {E x;while ( (x = unlinkFirst()) == null)notEmpty.await();return x;} finally {lock.unlock();}}public E takeLast() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lock();try {E x;while ( (x = unlinkLast()) == null)notEmpty.await();return x;} finally {lock.unlock();}}public E pollFirst(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {E x;while ( (x = unlinkFirst()) == null) {if (nanos <= 0)return null;nanos = notEmpty.awaitNanos(nanos);}return x;} finally {lock.unlock();}}public E pollLast(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {E x;while ( (x = unlinkLast()) == null) {if (nanos <= 0)return null;nanos = notEmpty.awaitNanos(nanos);}return x;} finally {lock.unlock();}}/*** @throws NoSuchElementException {@inheritDoc}*/public E getFirst() {E x = peekFirst();if (x == null) throw new NoSuchElementException();return x;}/*** @throws NoSuchElementException {@inheritDoc}*/public E getLast() {E x = peekLast();if (x == null) throw new NoSuchElementException();return x;}public E peekFirst() {final ReentrantLock lock = this.lock;lock.lock();try {return (first == null) ? null : first.item;} finally {lock.unlock();}}public E peekLast() {final ReentrantLock lock = this.lock;lock.lock();try {return (last == null) ? null : last.item;} finally {lock.unlock();}}

这些代码大同小异，都是使用 ReentrantLock 保证线程安全性。

使用 Condition 保证阻塞性。

个人启发

  1. 使用 ReentrantLock 保证线程安全性。可以说掌握这个就掌握了大部分的同步容器。

  2. 使用 Condition 保证阻塞性，掌握这个就掌握了大部分的阻塞队列容器。

  3. 所有的容器都有优缺点。比如双向队列，就有对应的并发容器。我们要学习原理，化为自己所用。

参考资料

https://blog.csdn.net/vernonzheng/article/details/8267541

https://blog.csdn.net/qq_38293564/article/details/80592429

LinkedBlockingDeque源码学习

目录

java多线程并发之旅-01-并发概览

这篇关于java多线程并发之旅-17--双端队列之 LinkedBlockingDeque的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/259372。

相关文章

Spring Boot @RestControllerAdvice全局异常处理最佳实践

Spring Boot @RestControllerAdvice全局异常处理最佳实践

《SpringBoot@RestControllerAdvice全局异常处理最佳实践》本文详解SpringBoot中通过@RestControllerAdvice实现全局异常处理,强调代码复用、统... 目录前言一、为什么要使用全局异常处理？二、核心注解解析1. @RestControllerAdvice2
阅读更多...
Spring IoC 容器的使用详解(最新整理)

Spring IoC 容器的使用详解(最新整理)

《SpringIoC容器的使用详解(最新整理)》文章介绍了Spring框架中的应用分层思想与IoC容器原理,通过分层解耦业务逻辑、数据访问等模块,IoC容器利用@Component注解管理Bean... 目录1. 应用分层2. IoC 的介绍3. IoC 容器的使用3.1. bean 的存储3.2. 方法注
阅读更多...
Spring事务传播机制最佳实践

Spring事务传播机制最佳实践

《Spring事务传播机制最佳实践》Spring的事务传播机制为我们提供了优雅的解决方案,本文将带您深入理解这一机制,掌握不同场景下的最佳实践,感兴趣的朋友一起看看吧... 目录1. 什么是事务传播行为2. Spring支持的七种事务传播行为2.1 REQUIRED（默认）2.2 SUPPORTS2
阅读更多...
怎样通过分析GC日志来定位Java进程的内存问题

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》：本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.
阅读更多...
Java进程异常故障定位及排查过程

Java进程异常故障定位及排查过程

《Java进程异常故障定位及排查过程》：本文主要介绍Java进程异常故障定位及排查过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、故障发现与初步判断1. 监控系统告警2. 日志初步分析二、核心排查工具与步骤1. 进程状态检查2. CPU 飙升问题3. 内存
阅读更多...
java中新生代和老生代的关系说明

java中新生代和老生代的关系说明

《java中新生代和老生代的关系说明》：本文主要介绍java中新生代和老生代的关系说明,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、内存区域划分新生代老年代二、对象生命周期与晋升流程三、新生代与老年代的协作机制1. 跨代引用处理2. 动态年龄判定3. 空间分
阅读更多...
Java设计模式---迭代器模式(Iterator)解读

Java设计模式---迭代器模式(Iterator)解读

《Java设计模式---迭代器模式(Iterator)解读》：本文主要介绍Java设计模式---迭代器模式(Iterator),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录1、迭代器（Iterator）1.1、结构1.2、常用方法1.3、本质1、解耦集合与遍历逻辑2、统一
阅读更多...
Java内存分配与JVM参数详解(推荐)

Java内存分配与JVM参数详解(推荐)

《Java内存分配与JVM参数详解(推荐)》本文详解JVM内存结构与参数调整,涵盖堆分代、元空间、GC选择及优化策略,帮助开发者提升性能、避免内存泄漏,本文给大家介绍Java内存分配与JVM参数详解,... 目录引言JVM内存结构JVM参数概述堆内存分配年轻代与老年代调整堆内存大小调整年轻代与老年代比例元空
阅读更多...
深度解析Java DTO(最新推荐)

深度解析Java DTO(最新推荐)

《深度解析JavaDTO(最新推荐)》DTO（DataTransferObject）是一种用于在不同层（如Controller层、Service层）之间传输数据的对象设计模式,其核心目的是封装数据,... 目录一、什么是DTO？DTO的核心特点：二、为什么需要DTO？（对比Entity）三、实际应用场景解析
阅读更多...
Java 线程安全与 volatile与单例模式问题及解决方案

Java 线程安全与 volatile与单例模式问题及解决方案

《Java线程安全与volatile与单例模式问题及解决方案》文章主要讲解线程安全问题的五个成因（调度随机、变量修改、非原子操作、内存可见性、指令重排序）及解决方案,强调使用volatile关键字... 目录什么是线程安全线程安全问题的产生与解决方案线程的调度是随机的多个线程对同一个变量进行修改线程的修改操
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2