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。

相关文章

SpringBoot使用Apache Tika检测敏感信息

SpringBoot使用Apache Tika检测敏感信息

《SpringBoot使用ApacheTika检测敏感信息》ApacheTika是一个功能强大的内容分析工具,它能够从多种文件格式中提取文本、元数据以及其他结构化信息,下面我们来看看如何使用Ap... 目录Tika 主要特性1. 多格式支持2. 自动文件类型检测3. 文本和元数据提取4. 支持 OCR（光学
阅读更多...
Java内存泄漏问题的排查、优化与最佳实践

Java内存泄漏问题的排查、优化与最佳实践

《Java内存泄漏问题的排查、优化与最佳实践》在Java开发中,内存泄漏是一个常见且令人头疼的问题,内存泄漏指的是程序在运行过程中,已经不再使用的对象没有被及时释放,从而导致内存占用不断增加,最终... 目录引言1. 什么是内存泄漏？常见的内存泄漏情况2. 如何排查 Java 中的内存泄漏？2.1 使用 J
阅读更多...
JAVA系统中Spring Boot应用程序的配置文件application.yml使用详解

JAVA系统中Spring Boot应用程序的配置文件application.yml使用详解

《JAVA系统中SpringBoot应用程序的配置文件application.yml使用详解》：本文主要介绍JAVA系统中SpringBoot应用程序的配置文件application.yml的... 目录文件路径文件内容解释1. Server 配置2. Spring 配置3. Logging 配置4. Ma
阅读更多...
Java 字符数组转字符串的常用方法

Java 字符数组转字符串的常用方法

《Java字符数组转字符串的常用方法》文章总结了在Java中将字符数组转换为字符串的几种常用方法,包括使用String构造函数、String.valueOf()方法、StringBuilder以及A... 目录1. 使用String构造函数1.1 基本转换方法1.2 注意事项2. 使用String.valu
阅读更多...
java脚本使用不同版本jdk的说明介绍

java脚本使用不同版本jdk的说明介绍

《java脚本使用不同版本jdk的说明介绍》本文介绍了在Java中执行JavaScript脚本的几种方式,包括使用ScriptEngine、Nashorn和GraalVM,ScriptEngine适用... 目录Java脚本使用不同版本jdk的说明1.使用ScriptEngine执行javascript2.
阅读更多...
Spring MVC如何设置响应

Spring MVC如何设置响应

《SpringMVC如何设置响应》本文介绍了如何在Spring框架中设置响应,并通过不同的注解返回静态页面、HTML片段和JSON数据,此外,还讲解了如何设置响应的状态码和Header... 目录1. 返回静态页面1.1 Spring 默认扫描路径1.2 @RestController2. 返回 html2
阅读更多...
Spring常见错误之Web嵌套对象校验失效解决办法

Spring常见错误之Web嵌套对象校验失效解决办法

《Spring常见错误之Web嵌套对象校验失效解决办法》：本文主要介绍Spring常见错误之Web嵌套对象校验失效解决的相关资料,通过在Phone对象上添加@Valid注解,问题得以解决,需要的朋... 目录问题复现案例解析问题修正总结  问题复现当开发一个学籍管理系统时，我们会提供了一个 API 接口去
阅读更多...
Java操作ElasticSearch的实例详解

Java操作ElasticSearch的实例详解

《Java操作ElasticSearch的实例详解》Elasticsearch是一个分布式的搜索和分析引擎,广泛用于全文搜索、日志分析等场景,本文将介绍如何在Java应用中使用Elastics... 目录简介环境准备1. 安装 Elasticsearch2. 添加依赖连接 Elasticsearch1. 创
阅读更多...
Spring核心思想之浅谈IoC容器与依赖倒置(DI)

Spring核心思想之浅谈IoC容器与依赖倒置(DI)

《Spring核心思想之浅谈IoC容器与依赖倒置(DI)》文章介绍了Spring的IoC和DI机制,以及MyBatis的动态代理,通过注解和反射,Spring能够自动管理对象的创建和依赖注入,而MyB... 目录一、控制反转 IoC二、依赖倒置 DI1. 详细概念2. Spring 中 DI 的实现原理三、
阅读更多...
SpringBoot 整合 Grizzly的过程

SpringBoot 整合 Grizzly的过程

《SpringBoot整合Grizzly的过程》Grizzly是一个高性能的、异步的、非阻塞的HTTP服务器框架,它可以与SpringBoot一起提供比传统的Tomcat或Jet... 目录为什么选择 Grizzly？Spring Boot + Grizzly 整合的优势添加依赖自定义 Grizzly 作为
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2