并发队列之LinkedBlockingQueue

  上一篇我们看了一下这个队列ConcurrentLinkedQueue,那就是一个无界非阻塞链表,我们这次来看看LinkedBlockingQueue,这个队列看名字就知道是一个阻塞式队列(也就是一个单向链表),基于独占锁实现的,比较简单;

一.LinkedBlockingQueue基本结构

  内部也是有一个Node类,下图所示,item存 实际数据,next指向下一个节点,一个有参构造器,没啥好说的;

  我们可以看看这个队列有的一些属性,其实大概能猜出来就是生产者消费者模型:

//队列实际容量
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();

  构造器中可以看到,默认最大数量就是65536个,虽然说也可以指定大小,我们一定程度上可以说这是一个有界阻塞队列;

//默认队列最大的数量就是65536个
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);
}
//也可以传一个实现Collection接口的类,比如List,然后遍历将其中的元素封装成节点丢到队列中去
//注意这里获取锁和释放锁
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();
    }
}

  我们简单的介绍了这个队列的基本结构,现在我们可以看看一些重要的方法; 

二.offer方法

  这个方法向队列最后添加一个元素,插入成功返回true,如果队列满了,就抛弃当前元素返回false;

public boolean offer(E e) {
    //如果为null,就直接抛错
    if (e == null) throw new NullPointerException();
    //count表示队列中实际数量
    final AtomicInteger count = this.count;
    //如果实际数量和队列最大容量相同,那么就不能再添加了,返回false
    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 {
        //队列没有满,就把新的节点丢进去,递增计数器,为什么这里要进行判断呢?上面不是进行判断了么?
        //还是由于并发,如果在上面那里判断容量之后,当前线程还没有获取锁,此时一个其他线程先获取锁然后执行offer方法然后释放锁,那么
        //这里需要再判断一次
        if (count.get() < capacity) {
            enqueue(node);        //注意getAndIncrement和incrementAndGet方法的区别,前者是返回自增之前的值,后者是返回自增之后的值
            c = count.getAndIncrement();
            //这里判断如果队列还没有满,就唤醒之前notFully条件队列中的线程,前面说过了notfull中存放的是什么线程
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        //释放锁
        putLock.unlock();
    }
    //如果c==0,表示队列中在添加节点之前就已经有一个节点了,就唤醒条件变量notEmpty中的线程,这些线程就会从队列中去取数据
    if (c == 0)
        signalNotEmpty();
    return c >= 0;
}

private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

三.put方法

  这个方法在队列尾部插入一个元素,如果队列有空闲则插入后直接返回,否则就阻塞当前线程直到队列空闲再插入;而且当前线程在阻塞的时候被其他线程调用了中断方法,就会抛异常;

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;
    //这个方法后面带有Interruptibly,说明当前线程获取锁后是可中断的
    putLock.lockInterruptibly();
    try {
        //节点数量到达了最大容量,就将当前线程放到条件变量notFull的队列中
        while (count.get() == capacity) {
            notFull.await();
        }
        //节点数量没有到最大,就在链表最后添加节点
        enqueue(node);
        //计数器加一,注意如果count等于4,那么c还是等于4,这个方法是原子自增,返回原来的值,注意和incrementAndGet方法的区别
        c = count.getAndIncrement();
        //这里如果c+1<capacity的话,说明队列还没有满,就唤醒notFull中的线程可以往队列中添加元素
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        //释放锁
        putLock.unlock();
    }
    //如果c==0表示队列中添加节点之前就已经有了一个节点,唤醒notEmpty中的线程去队列中取数据
    if (c == 0)
        signalNotEmpty();
}

private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

四.poll方法

  这个方法是从头部移除一个元素,如果队列为空就返回null,这个方法不阻塞;

public E poll() {
    //记录队列中节点数量
    final AtomicInteger count = this.count;
    //如果队列为空就返回null
    if (count.get() == 0)
        return null;
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    //获取锁
    takeLock.lock();
    try {
        //这里再判断一次也是为了防止在获取锁之前其他线程调用了poll方法取了节点
        //如果队列不为空,计数器减一
        if (count.get() > 0) {
            //删除第一个有数据的节点(由于第一个节点是哨兵节点,所以相当于删除的是第二个节点),方法实现在下面
            x = dequeue();
            c = count.getAndDecrement();
            //如果c>1说明移除头节点之后,队列不为空,就唤醒notEmpty中条件队列中的线程去队列中取数据
            if (c > 1)
                notEmpty.signal();
        }
    } finally {
        //释放锁
        takeLock.unlock();
    }
    //这里这个判断,注意一下在原子类AtomicInteger中两个方法,比如初始值为5,那么调用decrementAndGet方法返回的事4,
    //而调用getAndDecrement方法返回的是5,我们这里额c调用的是后者,所以表示删除节点之前队列的数量
    //所以这里的意思就是:如果删除队列之前的数量等于队列最大容量,那么删除之后队列肯定有空位置,于是就唤醒notFull条件队列中的线程
    //往队列中添加新的节点
    if (c == capacity)
        signalNotFull();
    return x;
}
//这个方法很简单稍微提一下,就是将第一个哨兵节点自己引用自己,更好的被gc收集
//将head指向第二个节点,取出该节点的值,然后将该节点内的item置为null,此节点就变成了一个哨兵节点
private E dequeue() {
    //删除头节点
    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;
}

private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

.peek方法

  这个方法获取队列头部元素但是不移除该节点;

public E peek() {
    //队列为空,返回null
    if (count.get() == 0)
        return null;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        //如果头节点的下一个节点是空,就返回null,因为此时头节点指的是哨兵节点啊
        Node<E> first = head.next;
        if (first == null)
            return null;
        else
        //头节点的下一个节点不为空,那么下一个节点肯定有数据,拿过来就行了
            return first.item;
    } finally {
        takeLock.unlock();
    }
}

六.take方法

  当前方法跟peek方法基本一样,只不过这个方法是阻塞的:从队列头删除一个节点,如果队列为空则阻塞当前线程直到队列不为空再执行操作,如果在阻塞的时候其他线程修改了中断标志,那么该线程就抛错;

//这个方法其实和poll方法基本一样,没什么好说的,注意可以抛异常
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;
}

七.remove方法

  删除队列中某个指定的元素,删除成功返回true,失败返回false;

public boolean remove(Object o) {
    //传入的是null,直接返回false
    if (o == null) return false;
    //获取两个锁,方法实现如下,在删除节点的时候,不能进行入队和出队操作,那些线程会被阻塞丢到AQS队列里
    fullyLock();
    try {
        //遍历找到对应的节点,删除掉,没找到返回false
        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();
    }
}

void fullyLock() {
    putLock.lock();
    takeLock.lock();
}
void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

//删除其实很容易,就是跟普通链表的删除一样,就是把当前要删除的节点前面的节点指向后面的节点
void unlink(Node<E> p, Node<E> trail) {
    p.item = null;
    trail.next = p.next;
    if (last == p)
        last = trail;
    //还是注意getAndDecrement方法返回的是减一之前的值,如果减一之后队列不是满的,就唤醒notFull中条件队列中的线程往队列中添加节点
    if (count.getAndDecrement() == capacity)
        notFull.signal();
}

  

八.总结

  我们简单的看了看LinkedBlockingQueue中一些比较重要的方法,比上一篇的ConcurrentLinkedQueue容易好多呀!

  其中ConcurrentLinkedQueue是无界非阻塞队列,底层是用单向链表实现,入队和出队使用CAS实现;而LinkedBlockingQueue是有界阻塞队列,底层是用单向链表实现,入队和出队分别用独占锁的方式去处理,所以入队和出队是可以同时进行的,而且还为两个独占锁配置了两个条件队列,用于存放被阻塞的线层,注意,这里涉及到好几个队列,一个是独占锁的AQS队列,一个是条件队列,一个是存放数据的队列,不要弄混淆了啊!

  用下面这个图强化记忆:

原文地址:https://www.cnblogs.com/wyq1995/p/12285317.html

时间: 2024-11-06 22:39:51

并发队列之LinkedBlockingQueue的相关文章

看秒杀系统的时候看到的关于并发队列的介绍,摘抄如下

并发队列的选择 Java的并发包提供了三个常用的并发队列实现,分别是:ArrayBlockingQueue.ConcurrentLinkedQueue 和 LinkedBlockingQueue  . ArrayBlockingQueue是初始容量固定的阻塞队列,我们可以用来作为数据库模块成功竞拍的队列,比如有10个商品,那么我们就设定一个10大小的数组队列. ConcurrentLinkedQueue使用的是CAS原语无锁队列实现,是一个异步队列,入队的速度很快,出队进行了加锁,性能稍慢. L

阻塞队列和并发队列

在并发编程中我们有时候需要使用线程安全的队列.要实现一个线程安全的队列有两种实现方式:一种是使用阻塞算法,另一种是使用非阻塞算法. 使用阻塞算法的队列可以用一个锁(入队和出队用同一把锁)或两个锁(入队和出队用不同的锁)等方式来实现,其中阻塞队列的典型是:BlockingQueue; 非阻塞的实现方式则可以使用循环CAS的方式来实现,非阻塞队列的典型例子是ConcurrentLinkedQueue. 注:并行与并发的区别: 1.并行是指两者同时执行一件事,比如赛跑,两个人都在不停的往前跑: 2.并

秒杀系统:并发队列 接口设计 并发请求数据安全处理

看秒杀系统的时候看到的关于并发队列的介绍,摘抄如下 并发队列的选择 Java的并发包提供了三个常用的并发队列实现,分别是:ArrayBlockingQueue.ConcurrentLinkedQueue 和 LinkedBlockingQueue  . ArrayBlockingQueue是初始容量固定的阻塞队列,我们可以用来作为数据库模块成功竞拍的队列,比如有10个商品,那么我们就设定一个10大小的数组队列. ConcurrentLinkedQueue使用的是CAS原语无锁队列实现,是一个异步

多线程编程:阻塞、并发队列的使用总结

最近,一直在跟设计的任务调度模块周旋,目前终于完成了第一阶段的调试.今天,我想借助博客园平台把最近在设计过程中,使用队列和集合的一些基础知识给大家总结一下,方便大家以后直接copy.本文都是一些没有技术含量的东西,只是做个总结,牛哥还请绕路. 老习惯,还是先跟各位纸上谈会儿兵,首先说说队列,他主要分为并发队列和阻塞队列,在多线程业务场景中使用最为普遍,我就主要结合我所做过的业务谈谈我对它们的看法,关于它们的API和官方解释就不提了. 并发队列 并发队列:最常见的业务场景就是多个线程共享同一个队列

解读 Java 并发队列 BlockingQueue

最近得空,想写篇文章好好说说 java 线程池问题,我相信很多人都一知半解的,包括我自己在仔仔细细看源码之前,也有许多的不解,甚至有些地方我一直都没有理解到位. 说到线程池实现,那么就不得不涉及到各种 BlockingQueue 的实现,那么我想就 BlockingQueue 的问题和大家分享分享我了解的一些知识. 本文没有像之前分析 AQS 那样一行一行源码分析了,不过还是把其中最重要和最难理解的代码说了一遍,所以不免篇幅略长.本文涉及到比较多的 Doug Lea 对 BlockingQueu

java并发包提供的三种常用并发队列实现

java并发包中提供了三个常用的并发队列实现,分别是:ConcurrentLinkedQueue.LinkedBlockingQueue和ArrayBlockingQueue. ConcurrentLinkedQueue使用的是CAS原语无锁队列实现,是一个异步队列,入队速度很快,出队进行了加锁,性能稍慢: LinkedBlockingQueue也是阻塞队列,入队和出队都用了加锁,当队空的时候线程会暂时阻塞:当队空的时候线程会暂时阻塞: ArrayBlockingQueue是初始容器固定的阻塞队

并发队列之ArrayBlockingQueue

上一篇我们说了并发队列中的LinkedBlockingQueue队列,这次我们看看ArrayBlockingQueue,看看名字,我们想象一下LinkedList和ArrayList的区别,我们可以知道ArrayBlockingQueue底层肯定是基于数组实现的,这是一个有界数组: ArrayBlockingQueue其中的组成部分和LinkedBlockingQueue及其相似,也是有两个条件变量,维护阻塞队列,实现了生产消费者模式: 一.简单认识ArrayBlockingQueue 先看看几

同步,异步,串行队列,并发队列,全局队列,主队列等概念的总结

同步,异步,串行队列,并发队列,全局队列,主队列等概念的总结 在GCD函数中, 我们常常碰到同步,异步,串行队列,并发队列,全局队列,主队列等概念,而这些概念又常常组合在一起, 十分头疼, 这篇文章就来梳理一下这些烦人的概念. 不想看长篇大论的, 直接看文章末尾的表格即可! 在此之前, GCD中还涉及到两个十分重要的概念, 就是任务和队列 任务(Task): 你需要执行的操作 队列(Queue): 存放任务的容器 GCD中两个重要的函数, 一个同步执行, 一个异步执行 dispatch_asyn

ios多线程操作(五)—— GCD串行队列与并发队列

GCD的队列可以分为2大类型,分别为串行队列和并发队列 串行队列(Serial Dispatch Queue): 一次只调度一个任务,队列中的任务一个接着一个地执行(一个任务执行完毕后,再执行下一个任务) 创建一个队列 dispatch_queue_t q = dispatch_queue_create(const char *label, dispatch_queue_attr_t attr) 参数: const char *label:队列的名称 dispatch_queue_attr_t