Java并发编程--并发队列原理之LinkedBlockingQueue

冷不防 2023-07-01 15:54 3阅读 0赞

### 文章目录 ###

*  LinkedBlockingQueue原理探究
 *   *   *  (1). 结构
         *  (2). LinkedBlockingQueue原理介绍
         *   *  1). offer操作
             *  2). put操作
             *  3). poll操作
             *  4). tack操作
             *  5). peek操作
             *  6). remove操作
         *  (3). 小结
     *

# LinkedBlockingQueue原理探究 #

LinkedBlockingQueue是使用独占锁实现的阻塞队列.

### (1). 结构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70]

有单向链表实现,有两个Node,分别用来存放首尾节点,还有一个原子变量Count用来记录队列元素个数.

还有两个ReentrantLock实例,分别用来空值元素入队和出队的原子性.

tackLock控制出队操作,putLock控制入队操作.

另外使用了两个条件变量,notEmpty(由tackLock锁获得,在出队是判断队列是否为空)和notFull(由putLock锁获得,在入队是判断队列是否已满).

### (2). LinkedBlockingQueue原理介绍 ###

#### 1). offer操作 ####

如果有空闲,插入元素并返回true.没有则返回false.

如果传入元素为null,则抛出异常

public boolean offer(E e) {
        
        // 传入元素为null抛出异常
        if (e == null) throw new NullPointerException();
        
        // 队列判满
        final AtomicInteger count = this.count;
        if (count.get() == capacity)// capacity默认为MAX_VALUE,可在构造中传参设置
            return false;
        
        // 构造新节点,获取入队锁对象
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();// 加锁
        try {
        
            // 如果队列不满,进队,并递增元素数
            if (count.get() < capacity) {
        
                // 将node节点链接到队列尾
                enqueue(node);
                // count自增1,并返回修改前的值
                c = count.getAndIncrement();
                // 如果添加后还有空间,唤醒之前条件阻塞的入队线程
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
        
            putLock.unlock();// 解锁
        }
        if (c == 0)
            // c为入队前队列中的元素数,c==0说明此时队列中至少有一个元素
            // 唤醒其他所有因为不能出队条件阻塞的线程
            signalNotEmpty();
        return c >= 0;
    }

#### 2). put操作 ####

向队列尾插入一个元素,如果队列有空闲则插入,队列已满就阻塞当前线程,直到队列有空闲插入成功后返回.

当被阻塞是其他线程设置了中断,抛出InterruptedExecption异常.

如果传入元素为null,抛出空指针异常.

public void put(E e) throws InterruptedException {
        
        // 判断传入元素是不是null
        if (e == null) throw new NullPointerException();
        
        // 构建新节点,获取入队锁对象和计数器
        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();
    }

#### 3). poll操作 ####

从队列头获取一个并移除一个元素,如果队列为空返回null.该方法并不等待其他线程入队元素.

public E poll() {
        
        // 队列为空返回null
        final AtomicInteger count = this.count;
        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();
                c = count.getAndDecrement();
                // 出队后队列不为空,则唤醒其他出队线程
                if (c > 1)
                    notEmpty.signal();
            }
        } finally {
        
            takeLock.unlock();
        }
        // 出队前队列满,则出队后队列有空隙,唤醒其他入队线程
        if (c == capacity)
            signalNotFull();
        return x;
    }

#### 4). tack操作 ####

获取队列的头部元素,并从队列中移除,如果队列为空,阻塞当前线程,直到队列不为空后返回元素.

该方法响应中断,会抛出异常.

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

#### 5). peek操作 ####

获取头部节点元素,但不移除.

加出队锁的目的是防止获取了元素,但其他线程在该方法返回前将它出队了,造成不一致,和空指针异常

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

#### 6). remove操作 ####

删除队列中指定的元素,有则删除,没有返回false.

public boolean remove(Object o) {
        
        // 删除null元素直接返回false
        if (o == null) return false;
        
        // 同时加入队锁和出队锁
        fullyLock();
        try {
        
            for (Node<E> trail = head, p = trail.next; p != null; trail = p, p = p.next) {
        
                // 找到目标节点,删除节点
                if (o.equals(p.item)) {
        
                    unlink(p, trail);
                    return true;
                }
            }
            return false;
        } finally {
        
            // 解入队锁和出队锁
            fullyUnlock();
        }
    }
    
    // 删除trail节点后的p节点
    void unlink(Node<E> p, Node<E> trail) {
        
        p.item = null;
        trail.next = p.next;
        // 如果p是尾节点,那么重新设置尾节点
        if (last == p)
            last = trail;
        // 如果当前队列满,删除元素后队列不满,唤醒入队线程
        if (count.getAndDecrement() == capacity)
            notFull.signal();
    }

### (3). 小结 ###

LinkedBlockingQueue是一个阻塞队列，内部由两个ReentrantLock来实现出入队列的线程安全，由各自的Condition对象的await和signal来实现等待和唤醒功能。它和ArrayBlockingQueue的不同点在于：

*  队列大小有所不同，ArrayBlockingQueue是有界的初始化必须指定大小，而LinkedBlockingQueue可以是有界的也可以是无界的(Integer.MAX\_VALUE)，对于后者而言，当添加速度大于移除速度时，在无界的情况下，可能会造成内存溢出等问题。
 *  数据存储容器不同，ArrayBlockingQueue采用的是数组作为数据存储容器，而LinkedBlockingQueue采用的则是以Node节点作为连接对象的链表。
 *  由于ArrayBlockingQueue采用的是数组的存储容器，因此在插入或删除元素时不会产生或销毁任何额外的对象实例，而LinkedBlockingQueue则会生成一个额外的Node对象。这可能在长时间内需要高效并发地处理大批量数据的时，对于GC可能存在较大影响。
 *  两者的实现队列添加或移除的锁不一样，ArrayBlockingQueue实现的队列中的锁是没有分离的，即添加操作和移除操作采用的同一个ReenterLock锁，而LinkedBlockingQueue实现的队列中的锁是分离的，其添加采用的是putLock，移除采用的则是takeLock，这样能大大提高队列的吞吐量，也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

##  ##

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200123153843142.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4,size_16,color_FFFFFF,t_70