LinkedBlockingQueue源码解析

时间  2019-12-14
标签 linkedblockingqueue 源码 解析 栏目 Java 繁體版
原文   原文链接

上一篇博客，咱们介绍了ArrayBlockQueue，知道了它是基于数组实现的有界阻塞队列，既然有基于数组实现的，那么必定有基于链表实现的队列了，没错，固然有，这就是咱们今天的主角：LinkedBlockingQueue。ArrayBlockQueue是有界的，那么LinkedBlockingQueue是有界仍是无界的呢？我以为能够说是有界的，也能够说是无界的，为何这么说呢？看下去你就知道了。node

和上篇博客同样，咱们仍是先看下LinkedBlockingQueue的基本应用，而后解析LinkedBlockingQueue的核心代码。数组

LinkedBlockingQueue基本应用

public static void main(String[] args) throws InterruptedException {
        LinkedBlockingQueue<Integer> linkedBlockingQueue = new LinkedBlockingQueue();

        linkedBlockingQueue.add(15);
        linkedBlockingQueue.add(60);
        linkedBlockingQueue.offer(50);
        linkedBlockingQueue.put(100);

        System.out.println(linkedBlockingQueue);

        System.out.println(linkedBlockingQueue.size());

        System.out.println(linkedBlockingQueue.take());
        System.out.println(linkedBlockingQueue);

        System.out.println(linkedBlockingQueue.poll());
        System.out.println(linkedBlockingQueue);

        System.out.println(linkedBlockingQueue.peek());
        System.out.println(linkedBlockingQueue);

        System.out.println(linkedBlockingQueue.remove(50));
        System.out.println(linkedBlockingQueue);
    }

运行结果：安全

[15, 60, 50, 100]
4
15
[60, 50, 100]
60
[50, 100]
50
[50, 100]
true
[100]

代码比较简单，先试着分析下：源码分析

  1. 建立了一个LinkedBlockingQueue 。
  2. 分别使用add/offer/put方法向LinkedBlockingQueue中添加元素，其中add方法执行了两次。
  3. 打印出LinkedBlockingQueue：[15, 60, 50, 100]。
  4. 打印出LinkedBlockingQueue的size：4。
  5. 使用take方法弹出第一个元素，并打印出来：15。
  6. 打印出LinkedBlockingQueue：[60, 50, 100]。
  7. 使用poll方法弹出第一个元素，并打印出来：60。
  8. 打印出LinkedBlockingQueue：[50, 100]。
  9. 使用peek方法弹出第一个元素，并打印出来：50。
  10. 打印出LinkedBlockingQueue：[50, 100]。
  11. 使用remove方法，移除值为50的元素，返回true。
  12. 打印出LinkedBlockingQueue：100。

代码比较简单，可是仍是有些细节不明白：this

  • 底层是如何保证线程安全性的?
  • 数据保存在哪里，以什么形式保存的?
  • offer/add/put都是往队列里面添加元素，区别是什么？
  • poll/take/peek都是弹出队列的元素，区别是什么？

要解决上面的疑问，最好的途径仍是看源码，下面咱们就来看看LinkedBlockingQueue的核心源码。线程

LinkedBlockingQueue源码解析

构造方法

LinkedBlockingQueue提供了三个构造方法，以下图所示：
image.png
咱们一个一个来分析。3d

LinkedBlockingQueue()
public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

无参的构造方法居然直接把“锅”甩出去了，甩给了另一个构造方法，可是咱们要注意传的参数：Integer.MAX_VALUE。指针

LinkedBlockingQueue(int capacity)
public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }
  1. 判断传入的capacity是否合法，若是不大于0，直接抛出异常。
  2. 把传入的capacity赋值给capacity。
  3. 新建一个Node节点，而且把此节点赋值给head和last字段。

这个capacity是什么呢？若是你们对代码有必定的感受的话，应该很容易猜到这是LinkedBlockingQueue的最大容量。若是咱们调用无参的构造方法来建立LinkedBlockingQueue的话，那么它的最大容量就是Integer.MAX_VALUE，咱们把它称为“无界”，可是咱们也能够指定最大容量，那么此队列又是一个“有界”队列了，因此有些博客很草率的说LinkedBlockingQueue是有界队列，或者是无界队列，我的认为这是不严谨的。code

咱们再来看看这个Node是个什么鬼：blog

static class Node<E> {
        E item;

        Node<E> next;

        Node(E x) { item = x; }
    }

是否是有一种莫名的亲切感，很明显，这是单向链表的实现呀，next指向的就是下一个Node。

LinkedBlockingQueue(Collection<? extends E> c)
public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);//调用第二个构造方法，传入的capacity是Int的最大值，能够说 是一个无界队列。
        final ReentrantLock putLock = this.putLock;
        putLock.lock(); //开启排他锁
        try {
            int n = 0;//用于记录LinkedBlockingQueue的size
            //循环传入的c集合
            for (E e : c) {
                if (e == null)//若是e==null，则抛出空指针异常
                    throw new NullPointerException();
                if (n == capacity)//若是n==capacity，说明到了最大的容量，则抛出“Queue full”异常
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));//入队操做
                ++n;//n自增
            }
            count.set(n);//设置count
        } finally {
            putLock.unlock();//释放排他锁
        }
    }
  1. 调用第二个构造方法，传入了int的最大值，因此能够说此时LinkedBlockingQueue是无界队列。
  2. 开启排他锁putLock 。
  3. 定义了一个变量n，用来记录当前LinkedBlockingQueue的size。
  4. 循环传入的集合，若是其中的元素为null，则抛出空指针异常，若是n==capacity，说明到了最大的容量，则抛出“Queue full”异常，不然执行enqueue操做来进行入队，而后n进行自增。
  5. 设置count为n，由此可知，count就是LinkedBlockingQueue的size了。
  6. 在finally中释放排他锁putLock 。

offer

public boolean offer(E e) {
        if (e == null) throw new NullPointerException();//若是传入的元素为NULL，抛出异常
        final AtomicInteger count = this.count;//取出count
        if (count.get() == capacity)//若是count==capacity，说明到了最大容量，直接返回false
            return false;
        int c = -1;//表示size
        Node<E> node = new Node<E>(e);//新建Node节点
        final ReentrantLock putLock = this.putLock;
        putLock.lock();//开启排他锁
        try {
            if (count.get() < capacity) {//若是count<capacity，说明尚未达到最大容量
                enqueue(node);//入队操做
                c = count.getAndIncrement();//得到count，赋值给c后完成自增操做
                if (c + 1 < capacity)//若是c+1 <capacity，说明还有剩余的空间，唤醒由于调用notFull的await方法而被阻塞的线程
                    notFull.signal();
            }
        } finally {
            putLock.unlock();//在finally中释放排他锁
        }
        if (c == 0)//若是c==0，说明释放putLock的时候，队列中有一个元素，则调用signalNotEmpty
            signalNotEmpty();
        return c >= 0;
    }
  1. 若是传进来的元素为null，则抛出异常。
  2. 把本类实例的count赋值给局部变量count。
  3. 若是count==capacity，说明到了最大的容量，直接返回false。
  4. 定义局部变量c，用来表示size，初始值是-1。
  5. 新建Node节点。
  6. 开启排他锁putLock。
  7. 若是count>=capacity，说明到了最大的容量，释放排他锁后，返回false，由于此时c=-1，c>=0为false；若是count<capacity，说明还有剩余空间，继续往下执行。这里须要思考一个问题，为何第三步已经判断过了是否还有剩余空间，这里还要再判断一次呢？由于可能有多个线程都在执行add/offer/put方法，当队列没有满的时候，多个线程同时执行到第三步（第三步的时候尚未开启排他锁），而后同时往下走，因此开启排他锁后，还须要从新判断下。
  8. 执行入队操做。
  9. 得到count，而且赋值给c后，完成自增的操做。注意，是先赋值后自增，赋值和自增的前后顺序会直接影响到后面的判断逻辑。
  10. 若是c+1<capacity，说明还有剩余的空间，唤醒由于调用notFull的await方法而被阻塞的线程。这里为何要+1再进行判断？由于在第9步中，是先赋值后自增，也就是说局部变量c保存的仍是入队以前LinkedBlockingQueue的size，因此要先进行+1操做，获得的才是当前LinkedBlockingQueue的size。
  11. 在finally中，释放排他锁putLock。
  12. 若是c==0，说明在释放putLock排他锁的时候，队列中有且只有一个元素，则调用signalNotEmpty方法。让咱们来看看signalNotEmpty方法：
private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

代码比较简单，就是开启排他锁，唤醒由于调用notEmpty的await方法而被阻塞的线程，可是这里须要注意，这里得到的排他锁已经再也不是putLock，而是takeLock。

add

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

add方法直接调用了offer方法，可是add和offer还不彻底同样，当队列满了，若是调用offer方法，会直接返回false，可是调用add方法，会抛出"Queue full"的异常。

put

public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();//若是传入的元素为NULL，抛出异常
        int c = -1;//表示size
        Node<E> node = new Node<E>(e);//新建Node节点
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;//得到count
        putLock.lockInterruptibly();//开启排他锁
        try {
            //若是到了最大容量，调用notFull的await方法，等待唤醒，用while循环，是为了防止虚假唤醒
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);//入队
            c = count.getAndIncrement();//count先赋值给c后，再进行自增操做
            if (c + 1 < capacity)//若是c+1<capacity，调用notFull的signal方法，唤醒由于调用notFull的await方法而被阻塞的线程
                notFull.signal();
        } finally {
            putLock.unlock();//释放排他锁
        }
        if (c == 0)//若是队列中有一个元素，唤醒由于调用notEmpty的await方法而被阻塞的线程
            signalNotEmpty();
    }
  1. 若是传入的元素为NULL，则抛出异常。
  2. 定义一个局部变量c，来表示size，初始值是-1。
  3. 新建Node节点。
  4. 把本类实例中的count赋值给局部变量count。
  5. 开启排他锁putLock。
  6. 若是到了最大容量，则调用notFull的await方法，阻塞当前线程，等待其余线程调用notFull的signal方法来唤醒本身，这里用while循环是为了防止虚假唤醒。
  7. 执行入队操做。
  8. count先赋值给c后，再进行自增操做。
  9. 若是c+1<capacity，说明还有剩余的空间，则调用notFull的signal方法，唤醒由于调用notFull的await方法而被阻塞的线程。
  10. 释放排他锁putLock。
  11. 若是队列中有且只有一个元素，唤醒由于调用notEmpty的await方法而被阻塞的线程。

enqueue

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

入队操做是否是特别简单，就是把传入的Node节点，赋值给last节点的next字段，再赋值给last字段，从而造成一个单向链表。

小总结

至此offer/add/put的核心源码已经分析完毕，咱们来作一个小总结，offer/add/put都是添加元素的方法，不过他们之间仍是有所区别的，当队列满了，调用以上三个方法会出现不一样的状况：

  • offer：直接返回false。
  • add：虽然内部也调用了offer方法，可是队列满了，会抛出异常。
  • put：线程会阻塞住，等待唤醒。

size

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

没什么好说的，count记录着LinkedBlockingQueue的size，得到后返回就是了。

take

public E take() throws InterruptedException {
        E x;
        int c = -1;//size
        final AtomicInteger count = this.count;//得到count
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();//开启排他锁
        try {
            while (count.get() == 0) {//说明目前队列中没有数据
                notEmpty.await();//阻塞，等待唤醒
            }
            x = dequeue();//出队
            c = count.getAndDecrement();//先赋值，后自减
            if (c > 1)//若是size>1，说明在出队以前，队列中有至少两个元素
                notEmpty.signal();//唤醒由于调用notEmpty的await方法而被阻塞的线程
        } finally {
            takeLock.unlock();//释放排他锁
        }
        if (c == capacity)//若是队列中还有一个剩余空间
            signalNotFull();
        return x;
    }
  1. 定义局部变量c，用来表示size，初始值是-1。
  2. 把本类实例的count字段赋值给临时变量count。
  3. 开启响应中断的排他锁takeLock 。
  4. 若是count==0，说明目前队列中没有数据，就阻塞当前线程，等待唤醒，直到其余线程调用了notEmpty的signal方法唤醒了当前线程。用while循环是为了防止虚假唤醒。
  5. 进行出队操做。
  6. count先赋值给c后，在进行自减操做，这里须要注意是先赋值，后自减。
  7. 若是c>1，也就是size>1，结合上面的先赋值，后自减，可知若是知足条件，说明在出队以前，队列中至少有两个元素，则调用notEmpty的signal方法，唤醒由于调用notEmpty的await方法而被阻塞的线程。
  8. 释放排他锁takeLock 。
  9. 若是执行出队后，队列中有且只有一个剩余空间，换个说法，就是执行出队操做前，队列是满的，则调用signalNotFull方法。

咱们再来看下signalNotFull方法：

private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }
  1. 开启排他锁，注意这里的排他锁是putLock 。
  2. 调用notFull的signal方法，唤醒由于调用notFull的await方法而被阻塞的线程。
  3. 释放排他锁putLock 。

poll

public E poll() {
        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;
    }

相比take方法，最大的区别就若是队列为空，执行take方法会阻塞当前线程，直到被唤醒，而poll方法，直接返回null。

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

peek方法，只是拿到头节点的值，可是不会移除该节点。

dequeue

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

没什么好说的，就是弹出元素，而且移除弹出的元素。

小总结

至此take/poll/peek的核心源码已经分析完毕，咱们来作一个小总结，take/poll/peek都是得到头节点值的方法，不过他们之间仍是有所区别的：

  • take：当队列为空，会阻塞当前线程，直到被唤醒。会进行出队操做，移除得到的节点。
  • poll：当队列为空，直接返回null。会进行出队操做，移除得到的节点。
  • peek：当队列为空，直接返回null。不会移除节点。

LinkedBlockingQueue的核心源码分析到这里完毕了，谢谢你们。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程