前言

本文收录于专辑:http://dwz.win/HjK,点击解锁更多数据结构与算法的常识。

你好,我是彤哥。

上一节,咱们一起学习了如何将递归改写为非递归,其中,用到的数据结构次要是栈。

栈和队列,能够说是除了数组和链表之外最根底的数据结构了,在很多场景中都有用到,前面咱们也会陆陆续续的看到。

明天,我想介绍一下,在Java中,如何构建一个高性能的队列,以及咱们须要把握的底层常识。

学习其余语言的同学,也能够看看,在你的语言中,是如何构建高性能队列的。

队列

队列,是一种先进先出(First In First Out,FIFO)的数据结构,相似于理论生存场景中的排队,先到的人先得。

应用数组和链表实现简略的队列,咱们后面都介绍过了,这里就不再赘述了,有趣味的同学能够点击以下链接查看:

重温四大根底数据结构:数组、链表、队列和栈

明天咱们次要来学习如何实现高性能的队列。

说起高性能的队列,当然是说在高并发环境下也可能工作得很好的队列,这里的很好次要是指两个方面:并发平安、性能好。

并发平安的队列

在Java中,默认地,也自带了一些并发平安的队列:

队列有界性数据结构
ArrayBlockingQueue有界加锁数组
LinkedBlockingQueue可选有界加锁链表
ConcurrentLinkedQueue无界无锁链表
SynchronousQueue无界无锁队列或栈
LinkedTransferQueue无界无锁链表
PriorityBlockingQueue无界加锁
DelayQueue无界加锁
这些队列的源码解析快捷入口:死磕 Java并发汇合之终结篇

总结起来,实现并发平安队列的数据结构次要有:数组、链表和堆,堆次要用于实现优先级队列,不具备通用性,暂且不探讨。

从有界性来看,只有ArrayBlockingQueue和LinkedBlockingQueue能够实现有界队列,其它的都是无界队列。

从加锁来看,ArrayBlockingQueue和LinkedBlockingQueue都采纳了加锁的形式,其它的都是采纳的CAS这种无锁的技术实现的。

从安全性的角度来说,咱们个别都要抉择有界队列,避免生产者速度过快导致内存溢出。

从性能的角度来说,咱们个别要思考无锁的形式,缩小线程上下文切换带来的性能损耗。

从JVM的角度来说,咱们个别抉择数组的实现形式,因为链表会频繁的增删节点,导致频繁的垃圾回收,这也是一种性能损耗。

所以,最佳的抉择就是:数组 + 有界 + 无锁。

而JDK并没有提供这样的队列,因而,很多开源框架都本人实现了高性能的队列,比方Disruptor,以及Netty中应用的jctools。

高性能队列

咱们这里不探讨具体的某一个框架,只介绍实现高性能队列的通用技术,并本人实现一个。

环形数组

通过下面的探讨,咱们晓得实现高性能队列应用的数据结构只能是数组,而数组实现队列,必然要应用到环形数组。

环形数组,个别通过设置两个指针实现:putIndex和takeIndex,或者叫writeIndex和readIndex,一个用于写,一个用于读。

当写指针达到数组尾端时,会从头开始,当然,不能越过读指针,同理,读指针达到数组尾端时,也会从头开始,当然,不能读取未写入的数据。

而为了避免写指针和读指针重叠的时候,无奈分清队列到底是满了还是空的状态,个别会再增加一个size字段:

所以,应用环形数组实现队列的数据结构个别为:

public class ArrayQueue<T> {    private T[] array;    private long wrtieIndex;    private long readIndex;    private long size;}

在单线程的状况下,这样不会有任何问题,然而,在多线程环境中,这样会带来重大的伪共享问题。

伪共享

什么是共享?

在计算机中,有很多存储单元,咱们接触最多的就是内存,又叫做主内存,此外,CPU还有三级缓存:L1、L2、L3,L1最贴近CPU,当然,它的存储空间也很小,L2比L1稍大一些,L3最大,能够同时缓存多个外围的数据。CPU取数据的时候,先从L1缓存中读取,如果没有再从L2缓存中读取,如果没有再从L3中读取,如果三级缓存都没有,最初会从内存中读取。离CPU外围越远,则绝对的耗时就越长,所以,如果要做一些很频繁的操作,要尽量保证数据缓存在L1中,这样能极大地提高性能。

缓存行

而数据在三级缓存中,也不是说来一个数据缓存一下,而是一次缓存一批数据,这一批数据又称作缓存行(Cache Line),通常为64字节。

每一次,当CPU去内存中拿数据的时候,都会把它前面的数据一并拿过去(组成64字节),咱们以long型数组为例,当CPU取数组中一个long的时候,同时会把后续的7个long一起取到缓存行中。

这在肯定水平上可能放慢数据的解决,因为,此时在解决下标为0的数据,下一个时刻可能就要解决下标为1的数据了,间接从缓存中取要快很多。

然而,这样又带来了一个新的问题——伪共享。

伪共享

试想一下,两个线程(CPU)同时在解决这个数组中的数据,两个CPU都缓存了,一个CPU在对array[0]的数据加1,另一个CPU在对array[1]的数据加1,那么,回写到主内存的时候,到底以哪个缓存行的数据为准(写回主内存的时候也是以缓存行的模式写回),所以,此时,就须要对这两个缓存行“加锁”了,一个CPU先批改数据,写回主内存,另一个CPU能力读取数据并批改数据,再写回主内存,这样势必会带来性能的损耗,呈现的这种景象就叫做伪共享,这种“加锁”的形式叫做内存屏障,对于内存屏障的常识咱们就不开展叙述了。

那么,怎么解决伪共享带来的问题呢?

以环形数组实现的队列为例,writeIndex、readIndex、size当初是这样解决的:

所以,咱们只须要在writeIndex和readIndex之间加7个long就能够把它们隔离开,同理,readIndex和size之间也是一样的。

这样就打消了writeIndex和readIndex之间的伪共享问题,因为writeIndex和readIndex必定是在两个不同的线程中更新,所以,打消伪共享之后带来的性能晋升是很显著的。

如果有多个生产者,writeIndex是必定会被争用的,此时,要怎么敌对地批改writeIndex呢?即一个生产者线程批改了writeIndex,另一个生产者线程要立马可见。

你第一工夫想到的必定是volatile,没错,可是光volatile还不行哦,volatile只能保障可见性和有序性,不能保障原子性,所以,还须要加上原子指令CAS,CAS是谁提供的?原子类AtomicInteger和AtomicLong都具备CAS的性能,那咱们间接应用他们吗?必定不是,仔细观察,发现他们最终都是调用Unsafe实现的。

OK,上面就轮到最牛逼的底层杀手退场了——Unsafe。

Unsafe

Unsafe不仅提供了CAS的指令,还提供很多其它操作底层的办法,比方操作间接内存、批改公有变量的值、实例化一个类、阻塞/唤醒线程、带有内存屏障的办法等。

对于Unsafe,能够看这篇文章:死磕 java魔法类之Unsafe解析

当然,构建高性能队列,次要应用的是Unsafe的CAS指令以及带有内存屏障的办法等:

// 原子指令public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);// 以volatile的模式获取值,相当于给变量加了volatile关键字public native long getLongVolatile(Object var1, long var2);// 提早更新,对变量的批改不会立刻写回到主内存,也就是说,另一个线程不会立刻可见public native void putOrderedLong(Object var1, long var2, long var4);

好了,底层常识介绍的差不多了,是时候展示真正的技术了——手写高性能队列。

手写高性能队列

咱们假如这样一种场景:有多个生产者(Multiple Producer),却只有一个消费者(Single Consumer),这是Netty中的经典场景,这样一种队列该怎么实现?

间接上代码:

/** * 多生产者单消费者队列 * * @param <T> */public class MpscArrayQueue<T> {    long p01, p02, p03, p04, p05, p06, p07;    // 寄存元素的中央    private T[] array;    long p1, p2, p3, p4, p5, p6, p7;    // 写指针,多个生产者,所以申明为volatile    private volatile long writeIndex;    long p11, p12, p13, p14, p15, p16, p17;    // 读指针,只有一个消费者,所以不必申明为volatile    private long readIndex;    long p21, p22, p23, p24, p25, p26, p27;    // 元素个数,生产者和消费者都可能批改,所以申明为volatile    private volatile long size;    long p31, p32, p33, p34, p35, p36, p37;    // Unsafe变量    private static final Unsafe UNSAFE;    // 数组根底偏移量    private static final long ARRAY_BASE_OFFSET;    // 数组元素偏移量    private static final long ARRAY_ELEMENT_SHIFT;    // writeIndex的偏移量    private static final long WRITE_INDEX_OFFSET;    // readIndex的偏移量    private static final long READ_INDEX_OFFSET;    // size的偏移量    private static final long SIZE_OFFSET;    static {        Field f = null;        try {            // 获取Unsafe的实例            f = Unsafe.class.getDeclaredField("theUnsafe");            f.setAccessible(true);            UNSAFE = (Unsafe) f.get(null);            // 计算数组根底偏移量            ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(Object[].class);            // 计算数组中元素偏移量            // 简略点了解,64位零碎中有压缩指针占用4个字节,没有压缩指针占用8个字节            int scale = UNSAFE.arrayIndexScale(Object[].class);            if (4 == scale) {                ARRAY_ELEMENT_SHIFT = 2;            } else if (8 == scale) {                ARRAY_ELEMENT_SHIFT = 3;            } else {                throw new IllegalStateException("未知指针的大小");            }            // 计算writeIndex的偏移量            WRITE_INDEX_OFFSET = UNSAFE                    .objectFieldOffset(MpscArrayQueue.class.getDeclaredField("writeIndex"));            // 计算readIndex的偏移量            READ_INDEX_OFFSET = UNSAFE                    .objectFieldOffset(MpscArrayQueue.class.getDeclaredField("readIndex"));            // 计算size的偏移量            SIZE_OFFSET = UNSAFE                    .objectFieldOffset(MpscArrayQueue.class.getDeclaredField("size"));        } catch (Exception e) {            throw new RuntimeException();        }    }    // 构造方法    public MpscArrayQueue(int capacity) {        // 取整到2的N次方(未思考越界)        capacity = 1 << (32 - Integer.numberOfLeadingZeros(capacity - 1));        // 实例化数组        this.array = (T[]) new Object[capacity];    }    // 生产元素    public boolean put(T t) {        if (t == null) {            return false;        }        long size;        long writeIndex;        do {            // 每次循环都从新获取size的大小            size = this.size;            // 队列满了间接返回            if (size >= this.array.length) {                return false;            }            // 每次循环都从新获取writeIndex的值            writeIndex = this.writeIndex;            // while循环中原子更新writeIndex的值            // 如果失败了从新走下面的过程        } while (!UNSAFE.compareAndSwapLong(this, WRITE_INDEX_OFFSET, writeIndex, writeIndex + 1));        // 到这里,阐明上述原子更新胜利了        // 那么,就把元素的值放到writeIndex的地位        // 且更新size        long eleOffset = calcElementOffset(writeIndex, this.array.length-1);        // 提早更新到主内存,读取的时候才更新        UNSAFE.putOrderedObject(this.array, eleOffset, t);        // 往死里更新直到胜利        do {            size = this.size;        } while (!UNSAFE.compareAndSwapLong(this, SIZE_OFFSET, size, size + 1));        return true;    }    // 生产元素    public T take() {        long size = this.size;        // 如果size为0,示意队列为空,间接返回        if (size <= 0) {            return null;        }        // size大于0,必定有值        // 只有一个消费者,不必思考线程平安的问题        long readIndex = this.readIndex;        // 计算读指针处元素的偏移量        long offset = calcElementOffset(readIndex, this.array.length-1);            // 获取读指针处的元素,应用volatile语法,强制更新生产者的数据到主内存        T e = (T) UNSAFE.getObjectVolatile(this.array, offset);        // 减少读指针        UNSAFE.putOrderedLong(this, READ_INDEX_OFFSET, readIndex+1);        // 减小size        do {            size = this.size;        } while (!UNSAFE.compareAndSwapLong(this, SIZE_OFFSET, size, size-1));        return e;    }    private long calcElementOffset(long index, long mask) {        // index & mask 相当于取余数,示意index达到数组尾端了从头开始        return ARRAY_BASE_OFFSET + ((index & mask) << ARRAY_ELEMENT_SHIFT);    }}

是不是看不懂?那就对了,多看几遍吧,面试又能吹一波了。

这里应用的是每两个变量之间加7个long类型的变量来打消伪共享,有的开源框架你可能会看到通过继承的形式实现的,还有的是加15个long类型,另外,JDK8中也提供了一个注解@Contended来打消伪共享。

本例其实还有优化的空间,比方,size的应用,能不能不应用size?不应用size又该如何实现?

后记

本节,咱们一起学习了在Java中如何构建高性能的队列,并学习了一些底层的常识,毫不夸大地讲,学会了这些底层常识,面试的时候光队列就能跟面试官吹一个小时。

另外,最近收到一些同学的反馈,说哈希、哈希表、哈希函数他们之间有关系吗?有怎么的关系?为什么Object中要放一个hash()办法?跟equals()办法怎么又扯上关系了呢?

下一节,咱们就来看看对于哈希的所有,想及时获取最新推文吗?还不快点来关注我!

关注公号主“彤哥读源码”,解锁更多源码、根底、架构常识。