BlockingQueue之LinkedBlockingQueue

电玩女神 2022-10-09 12:26 175阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

**BlockingQueue**的另一个实现类**LinkedBlockingQueue**，不同于**ArrayBlockingQueue**，LinkedBlockingQueue是基于链表实现的。*关于ArrayBlockingQueue的介绍，请戳《[BlockingQueue之ArrayBlockingQueue][BlockingQueue_ArrayBlockingQueue]》*

--------------------

# **█ Node** #

static class Node<E> {
        E item;
    
        Node<E> next;
    
        Node(E x) { item = x; }
    }

在LinkedBlockingQueue中维护了一个内部类**Node**，用于表示链表中的一个节点。其中**item**字段用来存放数据，**next**用来指向该节点的下一个节点。前一个节点通过next指向下一个节点，这样节点与节点之间相连就构成了链表。因为只有一个next指向下一个节点，因此在LinkedBlockingQueue中的链表是**单向链表**。

![20210624201957645.png][]

--------------------

# **█ 字段** #

private final int capacity;
    
    private final AtomicInteger count = new AtomicInteger();
    
    transient Node<E> head;
    
    private transient Node<E> last;
    
    private final ReentrantLock takeLock = new ReentrantLock();
    
    private final Condition notEmpty = takeLock.newCondition();
    
    private final ReentrantLock putLock = new ReentrantLock();
    
    private final Condition notFull = putLock.newCondition();

*  **capacity**

capacity用于记录当前LinkedBlockingQueue的队列的大小，即最多能存放capacity个元素。

*  **count**

count用来记录队列中已经存放的元素的个数，count有效范围介于0与capacity之间，即 0<=count<=capacity。当count=0时，表示队列为空，不存在任何元素。当count=capacity时，表示队列已满，没有空闲位置可以继续存放元素了。

*  **head**

LinkedBlockingQueue是基于链表实现的，前面介绍了Node类来表示链表上的每一个节点。但是还需要记录链表的起点吧，head就用来记录链表的起点，即链表上的第一个节点（头节点）。

*  **last**

与head相对，last用于记录链表上的最后一个节点（尾结点）。

*  **takeLock**

从队列中取出元素要加锁。

*  **putLock**

向队列中存放元素要加锁。

--------------------

# **█ 构造器** #

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);
    }

public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock(); // Never contended, but necessary for visibility
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

在构造器中初始化队列最大容量值**capacity**。使用空参构造器初始化的capacity大小为**Integer.MAX\_VALU**E，即队列中最多能存放**2147483647**个元素。在创建LinkedBlockingQueue对象的时候，建议最好不要使用空参构造器，应该手动指定队列的最大容量，因为当最大容量为Integer.MAX\_VALUE时，在某些情况下可能会因为队列中元素过多，导致内存溢出。

在构造器中还通过last = head = new Node(null)初始化了链表的头结点head与尾结点last，初始化时head与last节点中的数据item都为null，即是一个空节点。

![20210624202332798.png][]

--------------------

# **█ 方法** #

和学习ArrayBlockingQueue一样，学习LinkedBlockingQueue的方法也分两个部分，一部分是接口Queue中定义和约束的方法，一部分是BlockingQueue中定义和约束的方法。

**（1）Queue中定义的方法，不具有阻塞等待的行为。**

boolean add(E e);
        boolean offer(E e);
        E remove();
        E poll();
        E element();
        E peek();

*  **offer**

向队列中添加元素，队列满了无法继续添加，则返回false。

public boolean offer(E e) {
        // 不允许添加null，否则抛出空指针异常
        if (e == null) throw new NullPointerException();
        // 获取现在队列中元素的数量
        final AtomicInteger count = this.count;
        // 队列中的元素数量已经达到了队列容量了，无法继续添加元素
        if (count.get() == capacity)
            return false;
        int c = -1;
        // 将数据封装到Node节点对象中
        Node<E> node = new Node<E>(e);
        // 加锁，保证添加操作的安全性
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            // 队列还没满
            if (count.get() < capacity) {
                // 添加元素
                enqueue(node);
                // 获取count值后，再将count值增加1
                // c的值为count增加之前的旧值
                c = count.getAndIncrement();
                // 如果队列还没满，唤醒正在等待添加的线程
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }

关于第30行的代码，c==0的判断。c初始值为-1，c只有执行了c = count.getAndIncrement()以后值才能变化，而c取自count没增加1之前的值，因此需要count为0，则c才为0。也就是向队列中添加的是第一个元素，c==0才成立。

**enqueue：**

private void enqueue(Node<E> node) {
        last = last.next = node;
    }

向队列中添加元素。因为LinkedBlockingQueue的基于链表实现队列的，因此添加元素的操作就是将链表中最后一个的尾结点的next指针指向新的节点，并将last指向新的节点。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3][]

*  **add**

add方法继承自**AbstractQueue**。向队列中添加元素，当队列满了无法继续添加元素时，抛出异常**IllegalStateException**("Queue full")。

public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }

*  **poll**

从队列中取出元素。当队列为空没有元素可以取出时，返回null。

public E poll() {
        final AtomicInteger count = this.count;
        // count==0，表示队列为空，还没有元素
        if (count.get() == 0)
            return null;
        E x = null;
        int c = -1;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            if (count.get() > 0) {
                x = dequeue();
                // count数量减1
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

**dequeue：**

private E dequeue() {
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h;
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }

记住一点，head头结点始终为一个空的节点，头结点的下一个节点才是第一个元素的节点。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4][]

①Node h = head;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 5][]

②Node first = h.next;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 6][]

③h.next = h;

经过此步骤，原先的空节点就与整个链表断开了，变成了独立的一个节点。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 7][]

④head = first;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 8][]

⑤E x = first.item;

取出此时first节点的值，这个就是队列中的第一个值，出队。

⑥first.item = null;

将first节点的item设置成null，此时first节点就变成了链表的头空节点。原先的空节点因为与链表断开了，等待JVM垃圾回收。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 9][]

*  **remove**

remove方法继承自AbstractQueue。当队列为空没有元素可以移出时，会抛出异常**NoSuchElementException**。

public E remove() {
        E x = poll();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
    }

*  **peek**

查看队列中的第一个元素，第一个元素不存在时，返回null。在链表中，第一个节点永远是一个空的头节点，因此队列中的第一个元素是链表中的第二个节点所持有的值，因此是head.next。

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();
        }
    }

*  **element**

继承自AbstractQueue的方法。查看队列中的第一个元素，当第一个元素不存在时，抛出NoSuchElementException。

public E element() {
        E x = peek();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
    }

**（2）BlockingQueue中定义的方法，具有阻塞等待的行为。**

void put(E e) throws InterruptedException;
    
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
        
    E take() throws InterruptedException;
    
    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

*  **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>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

*  **offer**

向队列中添加元素，当队列满时，带有超时等待的功能。在超时时间达到以后如果还不能插入成功，则返回false。

public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
    
        if (e == null) throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                if (nanos <= 0)
                    return false;
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(new Node<E>(e));
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return true;
    }

*  **take**

从队列中取出一个元素。当队列为空没有元素可以取出的时候，会一直阻塞等待，直到队列中有了元素可以取出。

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;
    }

*  **poll**

从队列中取出一个元素，当队列为空时，带有超时等待。在超时时间达到以后，如果还取不出元素，则返回null。

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E x = null;
        int c = -1;
        long nanos = unit.toNanos(timeout);
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221005/260dc81b0db54a608cf8f12f7c5db8dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221005/dab98fc6c0d64c8982460cfb3a9b2ad4.png
[BlockingQueue_ArrayBlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/118162245?spm=1001.2014.3001.5501
[20210624201957645.png]: /images/20221005/736e0346034846e7b3800d431a4659c0.png
[20210624202332798.png]: /images/20221005/27387fd31e904f3181fa2baa2b9d8607.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20221005/1f65e238a147443287104dd529826acf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3]: /images/20221005/bf39789200844d72b8212ff784596b49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4]: /images/20221005/804101531e4e4a069e478f2284f58212.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 5]: /images/20221005/f045890b0c1c4bcf87d1b66c09c0b28c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 6]: /images/20221005/c0cae2fbdecc4ae0b3f2c46fc49d755b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 7]: /images/20221005/9fbf52ae47254ea7b791200797eca2ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 8]: /images/20221005/a2bf62d26f50445bb3ee9ada78d233a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 9]: /images/20221005/da40363854a748e196710185a84e52a0.png