LinkedBlockingDeque介绍
【1】LinkedBlockingDeque是一个基于链表实现的双向阻塞队列,默认情况下,该阻塞队列的大小为Integer.MAX_VALUE,可以看做无界队列,但也可以设置容量限制,作为有界队列。
【2】相比于其他阻塞队列,LinkedBlockingDeque 多了 addFirst、addLast、peekFirst、peekLast 等方法。以first结尾的方法,表示插入、获取或移除双端队列的第一个元素。以 last 结尾的方法,表示插入、获取或移除双端队列的最后一个元素。但本质上并没有优化锁的竞争情况,因为不管是从队首还是队尾,都是在竞争同一把锁,只不过数据插入和获取的方式多了。
LinkedBlockingDeque的源码分析
【1】属性值
//典型的双端链表结构 static final class Node<E> { E item; //存储元素 Node<E> prev; //前驱节点 Node<E> next; //后继节点 Node(E x) { item = x; } } // 链表头 本身是不存储任何元素的,初始化时item指向null transient Node<E> first; // 链表尾 transient Node<E> last; // 元素数量 private transient int count; // 容量,指定容量就是有界队列 private final int capacity; //重入锁 final ReentrantLock lock = new ReentrantLock(); // 当队列无元素时,锁会阻塞在notEmpty条件上,等待其它线程唤醒 private final Condition notEmpty = lock.newCondition(); // 当队列满了时,锁会阻塞在notFull上,等待其它线程唤醒 private final Condition notFull = lock.newCondition();
【2】构造函数
public LinkedBlockingDeque() { // 如果没传容量,就使用最大int值初始化其容量 this(Integer.MAX_VALUE); } public LinkedBlockingDeque(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; } public LinkedBlockingDeque(Collection<? extends E> c) { this(Integer.MAX_VALUE); final ReentrantLock lock = this.lock; lock.lock(); // 为保证可见性而加的锁 try { for (E e : c) { if (e == null) throw new NullPointerException(); //从尾部插入元素,插入失败抛出异常 if (!linkLast(new Node<E>(e))) throw new IllegalStateException("Deque full"); } } finally { lock.unlock(); } }
【3】核心方法分析
1)入队方法
//添加头结点元素 public void addFirst(E e) { //如果添加失败,抛出异常 if (!offerFirst(e)) throw new IllegalStateException("Deque full"); } //添加尾结点元素 public void addLast(E e) { //如果添加失败,抛出异常 if (!offerLast(e)) throw new IllegalStateException("Deque full"); } //添加头结点元素 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(); } } //添加尾结点元素 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(); } } //头部插入 private boolean linkFirst(Node<E> node) { //当前容量大于队列最大容量时,直接返回插入失败 if (count >= capacity) return false; //队列中的头结点 Node<E> f = first; //原来的头结点作为 新插入结点的后一个结点 node.next = f; //替换头结点 为新插入的结点 first = node; //尾结点不存在,将尾结点置为当前新插入的结点 if (last == null) last = node; else //原来的头结点的上一个结点为当前新插入的结点 f.prev = node; //当前容量增加 ++count; //唤醒读取时因队列中无元素而导致阻塞的线程 notEmpty.signal(); return true; } //尾部插入 private boolean linkLast(Node<E> node) { //当前容量大于队列最大容量时,直接返回插入失败 if (count >= capacity) return false; //获取尾节点 Node<E> l = last; //将新插入的前一个节点指向原来的尾节点 node.prev = l; //尾结点设置为新插入的结点 last = node; //头结点为空,新插入的结点作为头节点 if (first == null) first = node; else //将原尾结点的下一个结点指向新插入的节点 l.next = node; //当前容量增加 ++count; //唤醒读取时因队列中无元素而导致阻塞的线程 notEmpty.signal(); return true; } //头结点插入 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(); } } //尾结点插入 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(); } } //头结点插入 可指定阻塞时间 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(); } } //头结点插入 可指定阻塞时间 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; //获取可响应中断的锁,保证阻塞时间到期后可重新获得锁 try { //尾结点如果插入失败,会阻塞该方法,直到取出结点或删除结点时被唤醒 //或者阻塞时间到期直接返回失败 while (!linkLast(node)) { if (nanos <= 0) return false; nanos = notFull.awaitNanos(nanos); } return true; } finally { //解锁 lock.unlock(); } }
2)出队方法
//删除头结点 - 加锁 直接返回结果 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(); } } //删除头结点 - 加锁,如果删除失败则阻塞 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(); } } //删除头结点 private E unlinkFirst() { //获取当前头结点 Node<E> f = first; //头结点为空 返回空 if (f == null) return null; //获取头结点的下一个结点 Node<E> n = f.next; //获取头结点元素(记录return需要用到的删除了哪个元素) E item = f.item; //将头结点元素置为null f.item = null; //将头结点的下一个结点指向自己 方便gc f.next = f; //设置头结点为原头结点的下一个结点 first = n; //若原头结点的下一个结点不存在(队列中没有了结点) if (n == null) //将尾结点也置为null last = null; else //新的头结点的前一个结点指向null,因为他已经作为了头结点 所以不需要指向上一个结点 n.prev = null; //当前数量减少 --count; //唤醒因添加元素时队列容量满导致阻塞的线程 notFull.signal(); //返回原来的头结点中的元素 return item; } //删除尾结点 private E unlinkLast() { //获取当前尾结点 Node<E> l = last; //尾结点不存在 返回null if (l == null) return null; //获取当前尾结点的上一个结点 Node<E> p = l.prev; //获取当前尾结点中的元素,需要返回记录 E item = l.item; //将当前尾结点的元素置为null l.item = null; //并将当前尾结点的上一个结点指向自己,方便gc l.prev = l; //设置新的尾结点,为原来尾结点的前一个结点 last = p; //若无新的尾结点,头结点置为空(队列中没有了结点) if (p == null) first = null; else //将新的尾结点的下一个结点指向null,因为他已经为尾结点所以不需要指向下一个结点 p.next = null; //数量减少 --count; //唤醒因添加元素时队列容量满导致阻塞的线程 notFull.signal(); //返回原来的尾结点元素 return item; }
LinkedBlockingDeque总结
【1】一个链表阻塞双端无界队列,可以指定容量,默认为 Integer.MAX_VALUE
【2】数据结构:链表(同LinkedBlockingQueue,内部类Node存储元素)
【3】锁:ReentrantLock(同ArrayBlockingQueue)存取是同一把锁,操作的是同一个数组对象
【4】阻塞对象(notEmpty【出队:队列count=0,无元素可取时,阻塞在该对象上】,notFull【入队:队列count=capacity,放不进元素时,阻塞在该对象上】)
【5】入队,首尾都可以,均可以添加删除。
【6】出队,首尾都可以,均可以添加删除。
【7】应用场景:常用于 “工作窃取算法”。
标签:Node,node,结点,LinkedBlockingDeque,lock,阻塞,详解,null From: https://www.cnblogs.com/chafry/p/16782914.html