前言

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

你好，我是彤哥。

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

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

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

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

队列

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

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

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

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

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

并发平安的队列

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

这些队列的源码解析快捷入口：死磕 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()办法怎么又扯上关系了呢？

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

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