java源码分析09-LinkedBlockingQueue

拼搏现实的明天。 2022-08-17 14:26 10阅读 0赞

今天，我们来看看队列中的一员LinkedBlockingQueue的内部结构。

# 1.类结构 #

![Center][]

![Center 1][]

# 2. 类分析 #

## 2.1 类的元数据 ##

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {}

父类：AbstractQueue

接口：BlockingQueue，Serializable

## 2.2 类的属性 ##

/** The capacity bound, or Integer.MAX_VALUE if none */
        private final int capacity;
    
        /** Current number of elements */
        private final AtomicInteger count = new AtomicInteger(0);
    
        /**
         * Head of linked list.
         * Invariant: head.item == null
         */
        private transient Node<E> head;
    
        /**
         * Tail of linked list.
         * Invariant: last.next == null
         */
        private transient Node<E> last;
    
        /** Lock held by take, poll, etc */
        private final ReentrantLock takeLock = new ReentrantLock();
    
        /** Wait queue for waiting takes */
        private final Condition notEmpty = takeLock.newCondition();
    
        /** Lock held by put, offer, etc */
        private final ReentrantLock putLock = new ReentrantLock();
    
        /** Wait queue for waiting puts */
        private final Condition notFull = putLock.newCondition();

队列的容量

当前的数量，利用原子类来保证该属性的值不会因为多线程而出错。

头结点，尾结点

静态内部类

static class Node<E> {
            E item;
    
            /**
             * One of:
             * - the real successor Node
             * - this Node, meaning the successor is head.next
             * - null, meaning there is no successor (this is the last node)
             */
            Node<E> next;
    
            Node(E x) { item = x; }
        }

这是一个单向链表

tackLock与putLock分别表示从队列中取数据和放数据的时候，加的锁

notEmpty与notFull分别表示队列是否为空，是否满了

# 3.类的方法 #

## 3.1 put方法 ##

public void put(E e) throws InterruptedException {
            if (e == null) throw new NullPointerException();
            // Note: convention in all put/take/etc is to preset local var
            // holding count negative to indicate failure unless set.
            int c = -1;
            Node<E> node = new Node(e);
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
                /*
                 * Note that count is used in wait guard even though it is
                 * not protected by lock. This works because count can
                 * only decrease at this point (all other puts are shut
                 * out by lock), and we (or some other waiting put) are
                 * signalled if it ever changes from capacity. Similarly
                 * for all other uses of count in other wait guards.
                 */
                while (count.get() == capacity) {
                    notFull.await();
                }
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
        }

private void enqueue(Node<E> node) {
            // assert putLock.isHeldByCurrentThread();
            // assert last.next == null;
            last = last.next = node;
        }

首先，加入队列的元素不为空；其次，定义加元素锁和count，采用可中断锁lock来保持同步，先判断队列的元素个数是否已经达到容量上限，如果是那么put线程阻塞，否则，将该节点加入到单链表的尾部，并且队列的元素个数加一；此时需要判断

## 3.2 取元素 ##

public E take() throws InterruptedException {
            E x;
            int c = -1;
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lockInterruptibly();
            try {
                while (count.get() == 0) {
                    notEmpty.await();
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }

private E dequeue() {
            // assert takeLock.isHeldByCurrentThread();
            // assert head.item == null;
            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;
        }

首先定义取元素的锁以及队列大小，取之前加上可中断锁，先判断队列是否为空，为空则先等待；否则调用dequeue方法，取出head节点，重新调整单链表。

## 3.3 poll轮询 ##

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

先判断是否为空，为空返回null，

[Center]: /images/20220731/dcb7e0e0784c451486ca5888f3d65410.png
[Center 1]: /images/20220731/3cb55980523c4b2b8d897a2dad88896c.png