Java并发编程笔记之LinkedBlockingQueue源码探究

时间 2019-12-10

标签 java 并发编程笔记 linkedblockingqueue 源码探究栏目 Java 繁體版

原文原文链接

JDK 中基于链表的阻塞队列 LinkedBlockingQueue 原理剖析，LinkedBlockingQueue 内部是如何使用两个独占锁 ReentrantLock 以及对应的条件变量保证多线程先入队出队操做的线程安全？为何不使用一把锁，使用两把为什么能提升并发度？node

LinkedBlockingQueue的实现是使用独占锁实现的阻塞队列。首先看一下LinkedBlockingQueue 的类图结构，以下图所示：算法

如类图所示：LinkedBlockingQueue是使用单向链表实现，有两个Node分别来存放首尾节点，而且里面有个初始值为0 的原子变量count，它用来记录队列元素个数。安全

另外里面有两个ReentrantLock的实例，分别用来控制元素入队和出队的原子性，其中takeLock用来控制同时只有一个线程能够从队列获取元素，其余线程必须等待，多线程

putLock控制同时只能有一个线程能够获取锁去添加元素，其余线程必须等待。另外notEmpty 和 notFull 是信号量，内部分别有一个条件队列用来存放进队和出队的时候被阻塞的线程，并发

说白了，这其实就是一个生产者 - 消费者模型。函数

咱们首先看一下独占锁的源码，以下所示：高并发

　　/** 执行take, poll等操做时候须要获取该锁 */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** 当队列为空时候执行出队操做（好比take）的线程会被放入这个条件队列进行等待 */
    private final Condition notEmpty = takeLock.newCondition();

    /** 执行put, offer等操做时候须要获取该锁*/
    private final ReentrantLock putLock = new ReentrantLock();

    /**当队列满时候执行进队操做（好比put)的线程会被放入这个条件队列进行等待 */
    private final Condition notFull = putLock.newCondition();

 　　/** 当前队列元素个数 */
    private final AtomicInteger count = new AtomicInteger(0);

接着咱们要进入LinkedBlockingQueue 无参构造函数，源码以下：this

public static final int   MAX_VALUE = 0x7fffffff;

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

从源码中能够看到，默认队列的容量为0x7fffffff; 用户也能够本身指定容量，因此必定程度上 LinkedBlockingQueue 能够说是有界阻塞队列。spa

接下来咱们主要看LinkedBlockingQueue 的几个主要方法的源码，以下：线程

　　1.offer操做，向队列尾部插入一个元素，若是队列有空闲容量则插入成功后返回true，若是队列已满则丢弃当前元素而后返回false，若是 e元素为null，则抛出空指针异常(NullPointerException )，还有一点就是，该方法是非阻塞的。源码以下：

public boolean offer(E e) {

        //（1）空元素抛空指针异常
        if (e == null) throw new NullPointerException();

        //(2) 若是当前队列满了则丢弃将要放入的元素，而后返回false
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;

        //(3) 构造新节点，获取putLock独占锁
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            //(4)若是队列不满则进队列，并递增元素计数
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();
                //(5)
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            //(6)释放锁
            putLock.unlock();
        }
        //(7)
        if (c == 0)
            signalNotEmpty();
        //(8)
        return c >= 0;
}

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

代码（2）判断的是若是当前队列已满则丢弃当前元素并返回false。

代码（3）获取到putLock锁，当前线程获取到该锁后，则其余调用put 和 offer 的线程将会被阻塞（阻塞的线程被放到 putLock 锁的 AQS 阻塞队列）。

代码（4）这里又从新判断了一下当前队列是否满了，这是由于在执行代码（2）和获取到putLock锁期间，有可能其余线程经过put 或者 offer方法想队列里面添加了新的元素。从新判断队列确实不满则新元素入队，并递增计数器。

代码（5）判断的是若是新元素入队后还有空闲空间，则唤醒notFull的条件队列里面由于调用了notFull 的 await 操做（好比执行put方法而队列满了的时候）而被阻塞的一个线程，由于队列如今有空闲，因此这里能够提早唤醒一个入队线程。

代码（6）则释放获取的putLock锁，这里要注意锁的释放必定要在finally里面作，由于即便try块抛出异常了，finally也是会被执行到的。另外释放锁后其余由于调用put和offer而被阻塞的线程将会有一个获取到改锁。

代码（7）c == 0说明在执行代码（6）释放锁的时候队列里面至少有一个元素，队列里面有元素则执行signalNotEmpty，signalNotEmpty的源码以下：

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

经过上面代码能够看到其做用是激活notEmpty 的条件队列中由于调用notEmpty的await方法（好比调用 take 方法而且队列为空的时候）而被阻塞的一个线程，这里也说明了调用条件变量的方法前，要首先获取对应的锁。

offer的总结：offer方法中经过使用putLock锁保证了在队尾新增元素的原子性和队列元素个数的比较和递增操做的原子性。

　　2.put操做，向队列尾部插入一个元素，若是队列有空闲则插入后直接返回true,若是队列已经满则阻塞当前线程知道队列有空闲插入成功后返回true，若是在阻塞的时候被其余线程设置了中断标志，

则被阻塞线程会抛出InterruptedException 异常而返回，另外若是 e 元素为 null 则抛出 NullPointerException 异常。源码以下：

　　public void put(E e) throws InterruptedException {
        //（1）空元素抛空指针异常
        if (e == null) throw new NullPointerException();
        //(2) 构建新节点，并获取独占锁putLock
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            //(3)若是队列满则等待
            while (count.get() == capacity) {
                notFull.await();
            }
            //（4）进队列并递增计数
            enqueue(node);
            c = count.getAndIncrement();
            //(5)
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            //(6)
            putLock.unlock();
        }
        //(7)
        if (c == 0)
            signalNotEmpty();
    }

代码（2）中使用 putLock.lockInterruptibly() 获取独占锁，相比 offer 方法中这个获取独占锁方法意味着能够被中断，具体说是当前线程在获取锁的过程当中，若是被其它线程设置了中断标志则当前线程会抛出 InterruptedException 异常，

因此put操做在获取锁过程当中是可被中断的。

代码（3）若是当前队列已经满，则notFull 的 await() 把当前线程放入 notFull 的条件队列，当前线程被阻塞挂起并释放获取到的 putLock 锁，因为putLock锁被释放了，因此如今其余线程就有机会获取到putLock锁了。

代码（3）判断队列是否为空为什么使用 while 循环而不是 if 语句呢？

这是由于考虑到当前线程被虚假唤醒的问题，也就是其它线程没有调用 notFull 的 singal 方法时候，notFull.await() 在某种状况下会自动返回。

若是使用if语句简单判断一下，那么虚假唤醒后会执行代码（4），元素入队，而且递增计数器，而这时候队列已是满了的，致使队列元素个数大于了队列设置的容量，致使程序出错。

而使用使用 while 循环假如 notFull.await() 被虚假唤醒了，那么循环在检查一下当前队列是不是满的，若是是则再次进行等待。

　　3.poll操做，从队列头部获取并移除一个元素，若是队列为空则返回 null，该方法是不阻塞的。源码以下：

　　public E poll() {
        //(1)队列为空则返回null
        final AtomicInteger count = this.count;
        if (count.get() == 0)
            return null;
        //(2)获取独占锁
        E x = null;
        int c = -1;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            //(3)队列不空则出队并递减计数
            if (count.get() > 0) {//3.1
                x = dequeue();//3.2
                c = count.getAndDecrement();//3.3
                //(4)
                if (c > 1)
                    notEmpty.signal();
            }
        } finally {
            //(5)
            takeLock.unlock();
        }
        //(6)
        if (c == capacity)
            signalNotFull();
        //(7)返回
        return x;
    }
    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;
    }

代码(1) 若是当前队列为空，则直接返回 null。

代码（2）获取独占锁 takeLock，当前线程获取该锁后，其它线程在调用 poll 或者 take 方法会被阻塞挂起。

代码 (3) 若是当前队列不为空则进行出队操做，而后递减计数器。

代码（4）若是 c>1 则说明当前线程移除掉队列里面的一个元素后队列不为空（c 是删除元素前队列元素个数），那么这时候就能够激活由于调用 poll 或者 take 方法而被阻塞到notEmpty 的条件队列里面的一个线程。

代码（5）释放锁，必定要在finally里面释放锁。

代码（6）说明当前线程移除队头元素前当前队列是满的，移除队头元素后队列当前至少有一个空闲位置，那么这时候就能够调用signalNotFull激活由于调用put 或者 offer 而被阻塞放到 notFull 的条件队列里的一个线程，signalNotFull 源码以下：

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

poll 代码逻辑比较简单，值得注意的是获取元素时候只操做了队列的头节点。

　　4.peek 操做，获取队列头部元素可是不从队列里面移除，若是队列为空则返回 null，该方法是不阻塞的。源码以下：

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

能够看到代码（3）这里仍是须要判断下 first 是否为 null 的，不能直接执行代码（4）。

正常状况下执行到代码（2）说明队列不为空，可是代码（1）和（2）不是原子性操做，也就是在执行代码（1）判断队列不为空后，

在代码（2）获取到锁前，有可能其余线程执行了poll 或者 take 操做致使队列变为了空，而后当前线程获取锁后，直接执行 first.item 会抛出空指针异常。

　　5.take 操做，获取当前队列头部元素并从队列里面移除，若是队列为空则阻塞调用线程。若是队列为空则阻塞当前线程知道队列不为空，而后返回元素，若是在阻塞的时候被其余线程设置了中断标志，则被阻塞线程会抛出InterruptedException 异常而返回。源码以下：

　　public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        //(1)获取锁
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            //(2)当前队列为空则阻塞挂起
            while (count.get() == 0) {
                notEmpty.await();
            }
            //(3)出队并递减计数
            x = dequeue();
            c = count.getAndDecrement();
            //(4)
            if (c > 1)
                notEmpty.signal();
        } finally {
           //(5)
            takeLock.unlock();
        }
        //(6)
        if (c == capacity)
            signalNotFull();
        //(7)
        return x;
    }

代码（1）当前线程获取到独占锁，其余调用take 或者 poll的线程将会被阻塞挂起。

代码（2）若是队列为空则阻塞挂起当前线程，并把当前线程放入 notEmpty 的条件队列。

代码（3）进行出队操做并递减计数。

代码（4）若是 c > 1 说明当前队列不为空，则唤醒notEmpty 的条件队列的条件队列里面的一个由于调用 take 或者 poll 而被阻塞的线程。

代码（5）释放锁。

代码（6）若是 c == capacity 则说明当前队列至少有一个空闲位置，则激活条件变量 notFull 的条件队列里面的一个由于调用 put 或者 offer 而被阻塞的线程。

　　6.remove操做，删除队列里面指定元素，有则删除返回 true，没有则返回 false，源码以下：

public boolean remove(Object o) {
    if (o == null) return false;

    //（1）双重加锁
    fullyLock();
    try {

        //（2)遍历队列找则删除返回true
        for (Node<E> trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
             //(3)
            if (o.equals(p.item)) {
                unlink(p, trail);
                return true;
            }
        }
        //(4)找不到返回false
        return false;
    } finally {
        //(5)解锁
        fullyUnlock();
    }
}

代码（1）经过fullyLock获取双重锁，当前线程获取后，其余线程进行入队或者出队的操做就会被阻塞挂起。双重锁方法fullyLock的源码以下：

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

代码（2）遍历队列寻找要删除的元素，找不到则直接返回false,找到则执行unlink操做，unlink的源码以下：

　　void unlink(Node<E> p, Node<E> trail) {
      p.item = null;
      trail.next = p.next;
      if (last == p)
          last = trail;
      若是当前队列满，删除后，也不忘记唤醒等待的线程
      if (count.getAndDecrement() == capacity)
          notFull.signal();
    }

能够看到删除元素后，若是发现当前队列有空闲空间，则唤醒 notFull 的条件队列中一个由于调用 put 或者 offer 方法而被阻塞的线程。

代码（5）调用 fullyUnlock 方法使用与加锁顺序相反的顺序释放双重锁，源码以下：

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

　　7.size操做，获取当前队列元素个数。源码以下：

public int size() {
   return count.get();
}

总结：因为在操做出队入队的时候操做Count的时候加了锁，所以相比ConcurrentLinkedQueue 的size方法比较准确。

最后用一张图来加深LinkedBlockingQueue的理解，以下图：

所以咱们要思考一个问题：为什么 ConcurrentLinkedQueue 中须要遍历链表来获取 size 而不适用一个原子变量呢？

这是由于使用原子变量保存队列元素个数须要保证入队出队操做和操做原子变量是原子操做，而ConcurrentLinkedQueue 是使用 CAS 无锁算法的，因此没法作到这个。