转载

PriorityBlockingQueue 1.8 源码解析

[TOC]

PriorityBlockingQueue 1.8 源码解析

一，简介

PriorityBlockingQueue 是一个支持优先级的×××阻塞队列，数据结构采用的是 最小堆 是通过一个数组实现的，队列默认采用自然排序的升序排序，如果需要自定义排序，需要在构造队列时指定Comparetor比较器，队列也是使用ReentrantLock锁来实现的同步机制。

二，UML图

PriorityBlockingQueue 1.8 源码解析

三，基本成员

// 数组的最大容量 2^31 - 8
    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

     // 二叉堆数组
    private transient Object[] queue;

    // 总数
    private transient int size;

    // /默认比较器
    private transient Comparator<? super E> comparator;

    // 锁
    private final ReentrantLock lock;

    // 为空队列
    private final Condition notEmpty;

    // 自旋锁，在数组扩容时使用
    private transient volatile int allocationSpinLock;

注意：这里解释下这个Integer.MAX_VALUE - 8，为什么数组的最大长度是这么多了，这其实和int的最大值有关，最大值就是(1 << 32) -1 ,大家有没有发现数组的长度类型是int，为什么是int了？？？我也不知道，我也试了其它数据类型发现数组的长度必须是int类型，哈哈，所以也可以理解为什么是最大值了，至于为什么要减八了，是因为创建数组本身的信息（对象头，class信息啊）也是需要存储空间的，所以需要这8位的空间。

四，常用方法

入队方法

add 方法

public boolean add(E e) {
        return offer(e);
    }

put 方法

由于是×××队列所以put方法不会阻塞，也是直接调用了offer方法.

public void put(E e) {
        offer(e); // never need to block
    }

offer 带超时方法

public boolean offer(E e, long timeout, TimeUnit unit) {
        return offer(e); // never need to block
    }

offer 方法

// 添加元素
    public boolean offer(E e) {
        if (e == null)
            throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        int n, cap;
        Object[] array;
        // size大于等于数组的长度
        while ((n = size) >= (cap = (array = queue).length))
            // 扩容
            tryGrow(array, cap);
        try {
            Comparator<? super E> cmp = comparator;
            if (cmp == null) // 默认排序
                siftUpComparable(n, e, array);
            else // 自定义排序
                siftUpUsingComparator(n, e, array, cmp);
            size = n + 1;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

这里我们主要分析下offer方法里面的两个重要方法，扩容和入队，tryGrow，siftUpComparable方法。

tryGrow 方法

// 扩容方法
    private void tryGrow(Object[] array, int oldCap) {
        lock.unlock(); // must release and then re-acquire main lock
        Object[] newArray = null;
        // 只允许一个线程去扩容
        if (allocationSpinLock == 0 &&
                UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                        0, 1)) {
            try {
                // oldCap小于64  就加2 ，小于等于64就扩容50%
                int newCap = oldCap + ((oldCap < 64) ?
                        (oldCap + 2) : // grow faster if small
                        (oldCap >> 1));
                // 不可以超过MAX_ARRAY_SIZE
                if (newCap - MAX_ARRAY_SIZE > 0) {    // possible overflow
                    int minCap = oldCap + 1;
                    if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
                        throw new OutOfMemoryError();
                    newCap = MAX_ARRAY_SIZE;
                }
                if (newCap > oldCap && queue == array)
                    newArray = new Object[newCap];
            } finally {
                allocationSpinLock = 0;
            }
        }
        if (newArray == null) // back off if another thread is allocating
            Thread.yield(); // 扩容获取锁失败的线程，尽量让出cpu
        lock.lock(); // 重新获取锁
        if (newArray != null && queue == array) {
            queue = newArray;
            System.arraycopy(array, 0, newArray, 0, oldCap);
        }
    }

分析扩容：

  • lock.unlock()第一行为什么就要释放锁了，因为我们在入队之前就获取到了锁，如果我们不释放，那么别的线程无法入队也无法出队，这就大大降低了并发性，释放锁，别的线程就可以进行入队或者出队操作。
  • allocationSpinLock 这里为什么要用这样一个锁了，其实和我们第一行的释放锁有关系，我们在扩容时释放了锁，那就代表了其它线程也可以入队，但是队列满了，也需要扩容，所以这个锁就是为了让扩容只有一个线程来操作。
  • 在获取了allocationSpinLock 锁的线程在扩容中，我们发现其实只是创建了一个新的数组，并没有数据的迁移啥的，这是为什么了？？？后面再解释。
  • newArray == null ，其实就是别的线程来争夺扩容失败，然后尽量让出执行权（Thread.yield，线程从运行中变成就绪状态），让获取锁的线程去执行，但不是一定的，我们可以模拟一种可能，假设没有让出执行权，然后下一步获取到了锁。这时这个线程看见的newArray可能是null，所以就继续走offer方法的while ((n = size) >= (cap = (array = queue).length))循环，直到扩容线程完成对newArray 的改变。
  • lock.lock() 因为我们在扩容前释放了锁，允许别的线程对数组的操作，所以获取锁的一方面目的是为了控制只有一个线程对数组进行操作，第二个目的其实就是保证数组的可见性，别的线程可能在扩容期间执行了出队操作，保证下面数组的拷贝是准确的数据。

siftUpComparable 方法

最小堆的构建

// 保证了每条链的顺序小到大
    private static <T> void siftUpComparable(int k, T x, Object[] array) {
        Comparable<? super T> key = (Comparable<? super T>) x;
        while (k > 0) {
            int parent = (k - 1) >>> 1;
            Object e = array[parent];
            if (key.compareTo((T) e) >= 0)
                break;
            array[k] = e;
            k = parent;
        }
        array[k] = key;
    }

分析：

先得解释下(k - 1) >>> 1，就是求的商，我们来模拟插入五个数吧，默认容量是11.

  • 第一次插入一个1,此时的k是0，x是1，k不大于0，直接插入。

    索引 0
    值 1

  • 第二次我们插入一个0，此时的k是1，x是0，parent是0，然后获取0位置索引的值和现在的比较，现在其实是不大于0的，所以此时交换了位置，array[k] = e; k = parent;parent是0，所以结束循环然后在0的位置设置当前x是1。

    索引 0 1
    值 0 1

  • 第三次我们插入一个5，此时的k是2，x是5，parent 是0，然后获取0位置的值和插入值标记，发现是大于0的所以直接插入，在2的位置插入5。

    索引 0 1 2
    值 0 1 5

  • 第四次我们插入一个4，此时的k是3，x是4，parent是1，然后获取1位置的值和插入值比较，发现是大于0的，所以直接插入在3的位置插入。

    索引 0 1 2 3
    值 0 1 5 4

  • 第五次我们插入一个3，此时的k是4，x是3，parent是1，然后获取1位置值和插入值做比较，发现大于0的，所以直接在4的位置插入。

    索引 0 1 2 3 4
    值 0 1 5 4 3

我们用一个图来描绘下这个数组，怎么出现的这个图了，我们发现每次插入的数的索引就是数组的长度，然后通过（i - 1）>>> 2 = n求父节点，通过比较和父节点比较确认自己的位置，左右子节点其实就是2n+1,2n+2,左右子节点就是数组的相邻元素，我们发现子节点一定比父节点大，这就是最小堆；每次插入一个元素都是从最底层向上冒泡，维护最小堆的次序。

PriorityBlockingQueue 1.8 源码解析

出队方法

poll 方法

调用了 dequeue方法。

public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                // 弹出根节点
                return dequeue();
            } finally {
                lock.unlock();
            }
        }
    // 带超时时间
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        E result;
        try {
            while ( (result = dequeue()) == null && nanos > 0)
                nanos = notEmpty.awaitNanos(nanos);
        } finally {
            lock.unlock();
        }
        return result;
    }

take 方法

也是调用了dequeue方法，这个方法支持线程的中断。

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        E result;
        try {
            while ( (result = dequeue()) == null)
                notEmpty.await();
        } finally {
            lock.unlock();
        }
        return result;
    }

dequeue 方法

private E dequeue() {
        int n = size - 1;
        // 队列还没有初始化
        if (n < 0)
            return null;
        else {
            Object[] array = queue;
            // 获取根节点
            E result = (E) array[0];
            // 获取尾节点
            E x = (E) array[n];
            // 尾节点置位null
            array[n] = null;
            Comparator<? super E> cmp = comparator;
            // 重新排序最小堆
            if (cmp == null)
                siftDownComparable(0, x, array, n);
            else
                siftDownUsingComparator(0, x, array, n, cmp);
            size = n;
            return result;
        }
    }

其实上面就是返回了根节点，然后获取尾节点放在根节点的位置调整最小堆请看siftDownComparable方法。

siftDownComparable 方法

private static <T> void siftDownComparable(int k, T x, Object[] array,
                                               int n) {
        // n是数组的最大索引 k开始是0 x就是尾节点的值
        if (n > 0) {
            // x是最后一个节点的值
            Comparable<? super T> key = (Comparable<? super T>)x;
            int half = n >>> 1; // 最后一个节点的父节点           // loop while a non-leaf
            while (k < half) { // k是头节点 k> 了 说明到最后了
                int child = (k << 1) + 1; // assume left child is least // 左子节点
                Object c = array[child]; // 左节点的值
                int right = child + 1;  // 右子节点
                if (right < n && // 左子节点大于由子节点
                        ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                    c = array[child = right]; // c就是右子节点
                if (key.compareTo((T) c) <= 0) // 找到了子节点比自己大的
                    break;
                array[k] = c;
                k = child;
            }
            array[k] = key;
        }

    }

分析：

我们上图的5个元素为例，进行一次出队操作。

  • 弹出根节点0，然后调整最小堆。

PriorityBlockingQueue 1.8 源码解析

  • 我们调用siftDownComparable 方法调整最小堆，我们看下参数，此时的k是0，x是3，array就是这个数组，n就是4，key就是3，然后算half( half可以理解为堆中父节点最大索引位置，找到这个节点说明已经没有子节点了 )，half = 2。

  • 第一次，从0开始，此时的child是1，c=1，right是2，c>arrray[right],不大于，此时的key小于c不小于，所以此时0的位置就变成了1（array[k] = c），k = child。

PriorityBlockingQueue 1.8 源码解析

  • 第二次，从1开始，此时的child是3，c是4，right是4，左边大于右边所以，c=array[child=right ] = 3,所以此时的child 是4，c是3，由于key <= c，所以借宿循环，直接在1的位置设置c,调整结束。

PriorityBlockingQueue 1.8 源码解析

说下调整最小堆的过程，其实就是从根节点开始，重新构建父节点的过程，不过不是每个都需要重新构建，只需要构造子节点小的那边的的父节点，因为小的节点都去顶替原来的父节点了；我们弹出的是根节点，所以要从他的左右子节点找个根节点（但是要满足子节点大于父节点的规则），那么左右子节点有一个去当父节点了，它的位置也需要有节点代替，所以又从他的子节点开始找接替的节点，以此类推，直到找到最后一个父节点的位置。

size 方法

使用了锁，这个是精确的值。

public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return size;
        } finally {
            lock.unlock();
        }
    }

五，总结

PriorityBlockingQueue 是一个wujie的队列，使用put方法不会阻塞，使用时一定要注意内存溢出的问题；整个队列的出队和入队都是通过最小堆来实现的，理解最小堆是这个队列的关键；这个一个优先级的队列，适合有优先级的场景。

参考《Java 并发编程的艺术》

原文  https://blog.51cto.com/14220760/2414954
正文到此结束
所属分类： Java 编程技术

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 java-demo Maven教程 springcloud-demo hibernate教程 Docker 教程 memcached教程 Quartz指南 Hive教程 ANTLR教程 SpringCloud Ant教程 java实例教程 Hazelcast教程 XStream教程 Elastic-Job-Lite 深入浅出MyBatis ibaties教程 SVN教程 rabittmq教程 Hadoop教程 solr教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 使用 Docker 打包和运行 Vue 应用
  2. 2 Dependency Check命令行方式扫描jar包的安全漏洞
  3. 3 maven使用Dependency-Check来扫描安全漏洞
  4. 4 使用sonarqube分析代码质量
  5. 5 使用screw来对比数据库表和字段差异
  6. 6 Spring Boot 实现防盗链功能
  7. 7 在Spring Boot中的实现国际化（i18n）
  8. 8 一个用于 SMB/CIFS 网络的 Java 库-JCIFS
  9. 9 Spring Boot 3.x嵌入MongoDB 进行测试
  10. 10 java如何解析和生成yaml文件？
网站信息
  • 文章总数：82,737 篇
  • 文件总数：284,312 个
  • 标签总数：2,416 个
  • 分类总数：85 个
  • 留言数量：2,559 条
  • 在线人数：695 人
  • 运行天数：4,439天
  • 最后更新：2024年12月22日21点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG