4万字聊聊阿里二面，能抗住多少？

//无参构造ArrayList的话 会初始化空对象
private static final Object[] DEFAULTCAPACITY_EMPTY_ELEMENTDATA = {};
// 底层的数组对象
transient Object[] elementData;
// 默认初始化为10 
private static final int DEFAULT_CAPACITY = 10;
// 默认最大值
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
// 无参构造方法
public ArrayList() {
    //无惨构造，默认参数赋给数组对象
    this.elementData = DEFAULTCAPACITY_EMPTY_ELEMENTDATA;
}
// 有参构造方法
public ArrayList(int initialCapacity) {
    //有参构造时，会传一个大小，如果大于0 就把传入的参数作为length
    if (initialCapacity > 0) {
        this.elementData = new Object[initialCapacity];
    } else if (initialCapacity == 0) {
        //如果等于0，就直接用自带的DEFAULTCAPACITY_EMPTY_ELEMENTDATA
        this.elementData = EMPTY_ELEMENTDATA;
    } else {
        //如果这个大小小于0肯定是不允许的
        throw new IllegalArgumentException("Illegal Capacity: "+
                                           initialCapacity);
    }
}
// add方法
public boolean add(E e) {
    ensureCapacityInternal(size + 1);  // Increments modCount!!
    elementData[size++] = e;
    return true;
}
// add方法中的ensureCapacityInternal
private void ensureCapacityInternal(int minCapacity) {
    ensureExplicitCapacity(calculateCapacity(elementData, minCapacity));
}
// add方法中ensureCapacityInternal的calculateCapacity
// 这个函数其实就是处理 当是无参构造时，他会在add函数里 做+1处理 并且与自带的大小比较，取最大的值。
// 这里就是为什么很多会说 ArrayList初始化默认长度为10的原因（添加数据后是10）
private static int calculateCapacity(Object[] elementData, int minCapacity) {
    if (elementData == DEFAULTCAPACITY_EMPTY_ELEMENTDATA) {
        return Math.max(DEFAULT_CAPACITY, minCapacity);
    }
    return minCapacity;
}
// add方法中的ensureCapacityInternal的ensureExplicitCapacity
private void ensureExplicitCapacity(int minCapacity) {
    //modCount就是他的修改次数同时主要处理快速失败的一个作用，官方原话是这么说的
    //If an implementation does not wish to provide fail-fast iterators, this field may be ignored.
    //如果一个实现不希望提供快速失败迭代器，这个字段可以被忽略。  
    modCount++;
    // overflow-conscious code
    if (minCapacity - elementData.length > 0)
        grow(minCapacity);
}
// 这个函数主要处理的是一个 扩容 的一个作用。
// 扩容完之后， 开始处理数据的真实大小。通过Arrays.copyOf 函数 截取当前的 扩容后的大小 也就是15赋值给elementData 底层数组
private void grow(int minCapacity) {
        // 首先取整个链表 数据的长度，   比如 length = 10
        int oldCapacity = elementData.length;
     //10 + 5 = 15   这个 15 就是最新的链表长度 
        int newCapacity = oldCapacity + (oldCapacity >> 1);
        if (newCapacity - minCapacity < 0)
            newCapacity = minCapacity;
        if (newCapacity - MAX_ARRAY_SIZE > 0)
            newCapacity = hugeCapacity(minCapacity);
        // copy 数组，耗时主要是在这个地方
        elementData = Arrays.copyOf(elementData, newCapacity);
}
// 这个函数其实没啥用，主要就是两极的判断， 如果小于0 抛出异常，如果大于最大值 等等
private static int hugeCapacity(int minCapacity) {
  // 
     if (minCapacity < 0)
            throw new OutOfMemoryError();
        return (minCapacity > MAX_ARRAY_SIZE) ?
            Integer.MAX_VALUE :
            MAX_ARRAY_SIZE;
}
// 最后处理完之后， 回到add函数的 第二行代码，把新加的数据 写入当前数组对象中

JDK1.7与1.8的区别

1.7 默认构造大小为10

1.8 默认构造大小为空数组，长度为0

//无参构造方法   1.7
public ArrayList() {
        this(10);
}

 public ArrayList(int initialCapacity) {
        super();
        if (initialCapacity < 0)
            throw new IllegalArgumentException("Illegal Capacity: "+
                                               initialCapacity);
        this.elementData = new Object[initialCapacity];
}

//无参构造方法   1.8
public ArrayList() {
        this.elementData = DEFAULTCAPACITY_EMPTY_ELEMENTDATA;
}
private static final Object[] DEFAULTCAPACITY_EMPTY_ELEMENTDATA = {};

增删慢的原因是

在对源码进行分析的话，如果读懂源码的话，我相信大家对 grow 比较印象深刻。这个函数内会有一个Arrays.copyOf操作。也就是拷贝数组数据。这就是慢的根本原因。

tip: 这里LinkedList 我用的不多，就不做过多介绍了。后续会展开更新的再细一些。

②、哈希表（Hash）

哈希表（HashTable）也叫散列表，是一种可以通过关键码至 K,V直接访问的数据结构。它最大的特点就是可以快速实现查找、插入和删除。

数组的最大特点就是查找容易，插入和删除困难；而链表正好相反，查找困难，而插入和删除容易。哈希表很完美地结合了两者的优点， Java 的 HashMap 在此基础上还加入了树的优点。

上源码

final float loadFactor; //记录 hashMap 装载因子
static final float DEFAULT_LOAD_FACTOR = 0.75f; // 0.75
static final int MAXIMUM_CAPACITY = 1 << 30;  // 1073741824
static final int TREEIFY_THRESHOLD = 8;   //链表长度到8，就转为红黑树
static final int UNTREEIFY_THRESHOLD = 6; // 树大小为6，就转回链表
transient Node<K,V>[] table;  // 哈希桶数组
transient int modCount; //记录 hashMap 发生结构性变化的次数（注意 value 的覆盖不属于结构性变化）
int threshold; //threshold的值应等于 table.length * loadFactor, size 超过这个值时进行 resize()扩容
// 无参构造
public HashMap() {
    this.loadFactor = DEFAULT_LOAD_FACTOR; // all other fields defaulted
}
// 根据大小参数以及负载因子构造
public HashMap(int initialCapacity, float loadFactor) {
        if (initialCapacity < 0)
            throw new IllegalArgumentException("Illegal initial capacity: " +
                                               initialCapacity);
        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        if (loadFactor <= 0 || Float.isNaN(loadFactor))
            throw new IllegalArgumentException("Illegal load factor: " +
                                               loadFactor);
        this.loadFactor = loadFactor;
        this.threshold = tableSizeFor(initialCapacity);
}
// 根据大小参数构造   日常使用中，我们还是首选无参构造以及 根据大小参数构造的。
public HashMap(int initialCapacity) {
        this(initialCapacity, DEFAULT_LOAD_FACTOR);
}
// put 放入函数
public V put(K key, V value) {
    return putVal(hash(key), key, value, false, true);
}
// hash 函数
static final int hash(Object key) {
        int h;
    /*key 的 hash 值的计算是通过hashCode()的高16位异或低16位实现的：(h = k.hashCode()) ^ (h >>> 16)，主要是从速度、功效、质量来考虑的，这么做可以在数组table的length比较小的时候，也能保证考虑到高低Bit都参与到Hash的计算中，同时不会有太大的开销*/
        return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}
// put 函数的内嵌函数
// 参数onlyIfAbsent表示是否替换原值
// 参数evict我们可以忽略它，它主要用来区别通过put添加还是创建时初始化数据的
final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
                   boolean evict) {
        Node<K,V>[] tab; Node<K,V> p; int n, i;
     // 空表，需要初始化
        if ((tab = table) == null || (n = tab.length) == 0)
            // resize()不仅用来调整大小，还用来进行初始化配置
            n = (tab = resize()).length;
     //这里就是看下在hash位置有没有元素，实际位置是hash % (length-1)
        if ((p = tab[i = (n - 1) & hash]) == null)
            // 将元素直接插进去
            tab[i] = newNode(hash, key, value, null);
        else {
            //这时就需要链表或红黑树了
         // e是用来查看是不是待插入的元素已经有了，有就替换
            Node<K,V> e; K k;
            // p是存储在当前位置的元素
            if (p.hash == hash &&
                ((k = p.key) == key || (key != null && key.equals(k))))
                e = p;  //要插入的元素就是p，这说明目的是修改值
            // 若原来元素是红黑树节点，调用红黑树的插入方法:putTreeVal
            else if (p instanceof TreeNode)
                // 把节点添加到树中
                e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);
            else {
                // 这时候就是链表的头节点，要把待插入元素挂在链尾
                for (int binCount = 0; ; ++binCount) {
                    //向后循环
                    if ((e = p.next) == null) {
                        p.next = newNode(hash, key, value, null);
                        // 链表比较长，需要树化，
                     // 由于初始即为p.next，所以当插入第8个元素才会树化
                        if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
                            treeifyBin(tab, hash);
                        break;
                    }
                    // 找到了对应元素，就可以停止了
                    if (e.hash == hash &&
                        ((k = e.key) == key || (key != null && key.equals(k))))
                        break;
                    // 继续向后
                    p = e;
                }
            }
            // e就是被替换出来的元素，这时候就是修改元素值
            if (e != null) { // 待插入元素在 hashMap 中已存在
                V oldValue = e.value;
                if (!onlyIfAbsent || oldValue == null)
                    e.value = value;
                // 默认为空实现，允许我们修改完成后做一些操作
                afterNodeAccess(e);
                return oldValue;
            }
        }
        ++modCount;
     // size太大，达到了capacity的0.75，需要扩容
        if (++size > threshold)
            resize();
     // 默认也是空实现，允许我们插入完成后做一些操作
        afterNodeInsertion(evict);
        return null;
}
// 动态扩容
final Node<K,V>[] resize() {
        Node<K,V>[] oldTab = table;
     // 如果是第一次resize 肯定是0，无需扩容。 如果是装不下了  扩容的时候 就会选取 oldTab.length
        int oldCap = (oldTab == null) ? 0 : oldTab.length;
     // 判断负载因子是否大于0
        int oldThr = threshold;
        int newCap, newThr = 0;
     // 如果大于0 这里主要是做一个安全措施，如果大于了最大值MAXIMUM_CAPACITY  就会使用更大的上限
        if (oldCap > 0) {
            if (oldCap >= MAXIMUM_CAPACITY) {
                threshold = Integer.MAX_VALUE;
                return oldTab;
            }
           // 把上文的threshold 负载因子 扩大一倍  也就是 左移一位  赋值给newCap，与HashMap自带的上限做比较，并且与自带的默认容量比较  如果符合的话  就 计算当前的容量 （扩大一倍）
            else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY &&
                     oldCap >= DEFAULT_INITIAL_CAPACITY)
                newThr = oldThr << 1; // double threshold
        }
     //当前上文的负载因子大于0，把当前的值赋给 newCap （这个值主要是决定下文Node的大小）
        else if (oldThr > 0) // initial capacity was placed in threshold
            newCap = oldThr;
        else {               // zero initial threshold signifies using defaults
            // 否则就采用默认的大小 16
            newCap = DEFAULT_INITIAL_CAPACITY;
            newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY);
        }
     //这里主要就是判断 0 的一些保护措施
        if (newThr == 0) {
            float ft = (float)newCap * loadFactor;
            newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ?
                      (int)ft : Integer.MAX_VALUE);
        }
        threshold = newThr;
        @SuppressWarnings({"rawtypes","unchecked"})
     // 创建 Node的大小 ，把当前的Node节点给哈希桶数组
        Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
        table = newTab;
     // 判断oldTab  ，这里的oldTab其实就是 HashMap自带的 哈希桶数组在 resize第一行声明了
        if (oldTab != null) {
            // 开始对oldTab进行 循环， 终止值 就是 这个oldTab的大小 上文 判断了 取0 或者 .length了
            // 把 oldTab 中的节点 reHash 到 newTab 中去
            for (int j = 0; j < oldCap; ++j) {
                Node<K,V> e;
                // 每一次循环 把当前节点赋值给临时变量 e 判断是否为空
                if ((e = oldTab[j]) != null) {
                    // 当前节点归为null
                    oldTab[j] = null;
         // 如果当前只有一个节点，就取当前的hash值与 newCap-1 做二进制 位运算 直接在newTab中进行重定位
                    if (e.next == null)
                        newTab[e.hash & (newCap - 1)] = e;
                    // 如果当前 e 是树节点 要进行 红黑树的 rehash 操作
                    else if (e instanceof TreeNode)
                        ((TreeNode<K,V>)e).split(this, newTab, j, oldCap);
                    else { // 若是链表，进行链表的 rehash 操作
                        Node<K,V> loHead = null, loTail = null;
                        Node<K,V> hiHead = null, hiTail = null;
                        Node<K,V> next;
                        do {
                            next = e.next;
                            //根据算法 e.hash & oldCap 判断节点位置 rehash 后是否发生改变
                            if ((e.hash & oldCap) == 0) {
                                if (loTail == null)
                                    loHead = e;
                                else
                                    loTail.next = e;
                                loTail = e;
                            }
                            else {
                                if (hiTail == null)
                                    hiHead = e;
                                else
                                    hiTail.next = e;
                                hiTail = e;
                            }
                        } while ((e = next) != null);
                        if (loTail != null) {
                            loTail.next = null;
                            newTab[j] = loHead;
                        }
                        if (hiTail != null) {
                            hiTail.next = null;
                            // rehash 后节点新的位置一定为原来基础上加上 oldCap
                            newTab[j + oldCap] = hiHead;
                        }
                    }
                }
            }
        }
        return newTab;
}

 //这个函数的功能是对红黑树进行 rehash 操作
     final void split(HashMap<K,V> map, Node<K,V>[] tab, int index, int bit) {
             TreeNode<K,V> b = this;
             // Relink into lo and hi lists, preserving order
             TreeNode<K,V> loHead = null, loTail = null;
             TreeNode<K,V> hiHead = null, hiTail = null;
             int lc = 0, hc = 0;
          //由于 TreeNode 节点之间存在双端链表的关系，可以利用链表关系进行 rehash
             for (TreeNode<K,V> e = b, next; e != null; e = next) {
                    next = (TreeNode<K,V>)e.next;
                    e.next = null;
                    if ((e.hash & bit) == 0) {
                        if ((e.prev = loTail) == null)
                            loHead = e;
                        else
                            loTail.next = e;
                        loTail = e;
                        ++lc;
                    }
                    else {
                        if ((e.prev = hiTail) == null)
                            hiHead = e;
                        else
                            hiTail.next = e;
                        hiTail = e;
                        ++hc;
                    }
                }
                
                //rehash 操作之后注意对根据链表长度进行 untreeify 或 treeify 操作
                if (loHead != null) {
                    if (lc <= UNTREEIFY_THRESHOLD)
                        tab[index] = loHead.untreeify(map);
                    else {
                        tab[index] = loHead;
                        if (hiHead != null) // (else is already treeified)
                            loHead.treeify(tab);
                    }
                }
                if (hiHead != null) {
                    if (hc <= UNTREEIFY_THRESHOLD)
                        tab[index + bit] = hiHead.untreeify(map);
                    else {
                        tab[index + bit] = hiHead;
                        if (loHead != null)
                            hiHead.treeify(tab);
                    }
                }//end if
            }//end split

③、数组（Array）

这个Array可以参考上文的ArrayList。因为ArrayList的底层就是数组

④、栈（Stack）

提到栈我们的第一印象就是先进后出，下面我们看看栈的源码。栈的核心函数全部加了synchronized 锁，因为栈的数量，下标不能搞错，搞错就崩了。

protected int elementCount;
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
protected int capacityIncrement; // 如果容量增量小于或等于零，则每次需要增长时，向量的容量都会增加一倍。  
// 构造一个 无参的 栈。他继承于Vector
public Stack() {
}
// 入栈
public E push(E item) {
        addElement(item);

        return item;
}
public synchronized void addElement(E obj) {
    // 入栈时  递增 modCount 记录修改次数
        modCount++;
        ensureCapacityHelper(elementCount + 1);
        elementData[elementCount++] = obj;
}
private void ensureCapacityHelper(int minCapacity) {
        // 如果
        if (minCapacity - elementData.length > 0)
            grow(minCapacity);
}
// 这个函数应该在ArrayLisrt会比较熟悉的， 
private void grow(int minCapacity) {
        // overflow-conscious code
        int oldCapacity = elementData.length;
     // 如果当前容量 大于0 说明不需要增长  直接取当前容量， 否则就取  整个容量的 2倍（这里的2倍主要是从 oldCapacity+oldCapacity 得到的 oldCapacity又代表elementData.length ）
        int newCapacity = oldCapacity + ((capacityIncrement > 0) ?
                                         capacityIncrement : oldCapacity);
    // 这里就是一些安全措施
        if (newCapacity - minCapacity < 0)
            newCapacity = minCapacity;
        if (newCapacity - MAX_ARRAY_SIZE > 0)
            newCapacity = hugeCapacity(minCapacity);
    // newCapacity 就是 上文我们 三目运算符计算的 进行对象复制
        elementData = Arrays.copyOf(elementData, newCapacity);
}
// 出栈
public synchronized E pop() {
        E       obj;
        int     len = size();

        obj = peek();
        removeElementAt(len - 1);

        return obj;
}

// 这里直接返回容量的大小
public synchronized int size() {
        return elementCount;
}
public synchronized E peek() {
        int     len = size();

        if (len == 0)
            throw new EmptyStackException();
        return elementAt(len - 1);
}
// 由上列调用可以看出，取的就是末尾数据 弹出
public synchronized E elementAt(int index) {
    // 这里做了安全措施，如果当前弹出的下标 大于 容量  就抛出异常。 这句代码也验证了 顶部的 ` 因为栈的数量，下标不能搞错， 搞错就崩了。`
        if (index >= elementCount) {
            throw new ArrayIndexOutOfBoundsException(index + " >= " + elementCount);
        }
    // 弹出当前下标的元素  给上文的 obj
    return elementData(index);
}

public synchronized void removeElementAt(int index) {
     // 每次 删除的时候  自增一个 记录修改次数
        modCount++;
     // 如果当前的下标 大于 容量  抛出异常   （这个就是一个安全措施 没啥好说的）
        if (index >= elementCount) {
            throw new ArrayIndexOutOfBoundsException(index + " >= " +
                                                     elementCount);
        }
        else if (index < 0) {
            throw new ArrayIndexOutOfBoundsException(index);
        }
        int j = elementCount - index - 1;
     // 复制数组
        if (j > 0) {
            System.arraycopy(elementData, index + 1, elementData, index, j);
        }
     // 复制成功后  数量-1
        elementCount--;
     // 弹出的那个对象 为null 清空回收 结束
        elementData[elementCount] = null; /* to let gc do its work */
}

⑤、队列（Queue）

Queue继承于Collection 包，先进先出。它只是一个接口。旗下大概有6种实现类

ArrayBlockingQueue ：一个由数组支持的有界队列。
LinkedBlockingQueue ：一个由链接节点支持的可选有界队列。
PriorityBlockingQueue ：一个由优先级堆支持的无界优先级队列。
DelayQueue ：一个由优先级堆支持的、基于时间的调度队列。
SynchronousQueue ：一个利用 BlockingQueue 接口的简单聚集（rendezvous）机制。

ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(3);
//  这里大概的看了一下 感觉都很像，其实好好想想 这三种 都是继承AbstractQueue 而且又实现了Queue 能不像嘛
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
}
LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue();
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);
}
PriorityBlockingQueue priorityBlockingQue = new PriorityBlockingQueue();
public PriorityBlockingQueue() {
        this(DEFAULT_INITIAL_CAPACITY, null);
}
public PriorityBlockingQueue(int initialCapacity,
                                 Comparator<? super E> comparator) {
        if (initialCapacity < 1)
            throw new IllegalArgumentException();
        this.lock = new ReentrantLock();
        this.notEmpty = lock.newCondition();
        this.comparator = comparator;
        this.queue = new Object[initialCapacity];
}

这里主要就分析一下第一个队列吧

ArrayBlockingQueue是GUC包下的一个线程安全的阻塞队列，底层使用数组实现。

// 底层存储元素的数组
final Object[] items;
// 出队序号，如果有一个元素出队，那么后面的元素不会向前移动，
// 而是将 takeIndex + 1 表示后面要出队的元素的角标
int takeIndex;
// 入队序号，表示后续插入的元素的角标，putIndex 不一定大于 takeIndex
int putIndex;
// 元素个数
int count;
// 内部锁
final ReentrantLock lock;
// notEmpty 条件
private final Condition notEmpty;
// notFull 条件
private final Condition notFull;

// 这是一个有参构造，因为底层是数组实现的，所以肯定是要初始化数组的，数组的长度肯定是你传入的参数。
// 因为他是线程安全的，所以肯定也是少不了锁的存在。
// 锁这个东西这里就不展开了在下文中也会多次提到GUC的。
public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        // 初始化底层数组
        this.items = new Object[capacity];
        // 默认为非公平锁 非公平锁就是 false  代表 不会按照等待越长就越先执行的规则。 
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
}
public void put(E e) throws InterruptedException {
        checkNotNull(e);
     // 获取锁
        final ReentrantLock lock = this.lock;
     /**
         * lock：调用后一直阻塞到获得锁
         * tryLock：尝试是否能获得锁 如果不能获得立即返回
         * lockInterruptibly：调用后一直阻塞到获得锁 但是接受中断信号(比如：Thread、sleep)
         */
        lock.lockInterruptibly();
        try {
            // 如果队列数组已满，则 notFull 阻塞，当有元素被移除后才唤醒 notFull
            while (count == items.length)
                notFull.await();
            // 元素入队
            enqueue(e);
        } finally {
            // 添加完元素后释放锁，这里一定要注意，千万不要因为忘记放锁，导致程序超时
            lock.unlock();
        }
}
// 首先肯定是要判断是否为空， 如果为空的话抛出异常，这个属于安全措施 没啥好说的
private static void checkNotNull(Object v) {
        if (v == null)
            throw new NullPointerException();
}
// 插入元素的函数，只在加锁状态下调用  而且是插入在尾部
private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
    // 要插入的元素e 放入 items数组的 putIndex的位置  （putIndex 就是当前最新要放入的位置下标）
        items[putIndex] = x;
    // 先自增 再匹配 整个数组的长度  这里的写法真漂亮。 没有无效代码
        if (++putIndex == items.length)
            // 下次放入的位置为0  因为是先进先出嘛，所以就导致了 每次放入的位置都是0
            putIndex = 0;
     //元素个数
        count++;
      // 唤醒 notFull
        notEmpty.signal();
}

private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        // 元素置 null
        items[takeIndex] = null;
        // 如果出队的是数组中的最后一个元素，则重置 takeIndex 为 0
        if (++takeIndex == items.length)
            takeIndex = 0;
     // 出队一个 肯定是要递减的呀
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        // 唤醒 notFull
        notFull.signal();

搞完队列，栈啥的才发现好简单，HashMap的源码是真的复杂，不过学到的东西也是特别多的。

⑥、图（Graph）

不是特别常用

⑦、堆（Heap）

⑧、树（Tree）

TreeMap
TreeSet

这里主要介绍一下TreeSet，因为TreeSet是在TreeMap的基础上扩展出的一个有序集合，底层的存储也是一颗红黑树。

TreeMap<Integer,String> tree = new TreeMap<Integer,String>();
//定义NavigableMap接口类型m
private transient NavigableMap<E,Object> m;
//申请类型为Object的PRESENT，作为key-value值中的value值，将我们需要保存的值放入key中
private static final Object PRESENT = new Object();
// The comparator used to maintain order in this tree map, or null if it uses the natural ordering of its keys.
private final Comparator<? super K> comparator;

//多态，m其实为一个TreeMap的实现类
TreeSet(NavigableMap<E,Object> m) {
    this.m = m;
}
//无参构造函数，申请一个TreeMap作为参数，由此可见TreeSet依赖于TreeMap进行实现
public TreeSet() {
    this(new TreeMap<E,Object>());
}
//指定比较器的构造函数
public TreeSet(Comparator<? super E> comparator) {
    this(new TreeMap<>(comparator));
}
// 添加函数 源码
public V put(K key, V value) {
    // 声明一个 主节点 root
        Entry<K,V> t = root;
    // 如果是第一次插入的话 肯定是空，因为他没有父亲，他的主节点就是他自己
        if (t == null) {
            //使用此TreeMap的正确比较方法比较两个键。   下午已经把源码贴出来了
            compare(key, key); // type (and possibly null) check
            // 创建一个父节点
            root = new Entry<>(key, value, null);
            // 因为只有一个节点 size设为 1
            size = 1;
            modCount++;
            return null;
        }
     // 如果走到了这里说明 root 不为空，当前存在节点
        int cmp;
     // 声明一个临时 父亲节点
        Entry<K,V> parent;
        // split comparator and comparable paths
        Comparator<? super K> cpr = comparator;
        if (cpr != null) {
            do {
                // 当前比较的两个节点 不为空时， 就把之前的root 覆给 临时父节点
                parent = t;
                // 比较当前的key值的顺序
                cmp = cpr.compare(key, t.key);
                if (cmp < 0)
                    t = t.left;
                else if (cmp > 0)
                    t = t.right;
                else
                    return t.setValue(value);
            } while (t != null);
        }
        else {
            if (key == null)
                throw new NullPointerException();
            @SuppressWarnings("unchecked")
                Comparable<? super K> k = (Comparable<? super K>) key;
            do {
                parent = t;
                cmp = k.compareTo(t.key);
                if (cmp < 0)
                    t = t.left;
                else if (cmp > 0)
                    t = t.right;
                else
                    return t.setValue(value);
            } while (t != null);
        }
        Entry<K,V> e = new Entry<>(key, value, parent);
        if (cmp < 0)
            parent.left = e;
        else
            parent.right = e;
        fixAfterInsertion(e);
        size++;
        modCount++;
        return null;
}
// Compares two keys using the correct comparison method for this TreeMap.
// 使用此TreeMap的正确比较方法比较两个键。  
final int compare(Object k1, Object k2) {
        return comparator==null ? ((Comparable<? super K>)k1).compareTo((K)k2)
            : comparator.compare((K)k1, (K)k2);
}

树是有点麻烦了

1.3 JVM的双亲委派机制说一下

class文件是通过类加载器（ClassLoader）装载到JVM中的。为了防止存在多个字节码，使用了双亲委派机制。双亲委派机制一定分为3/4层

根加载器（Bootstrap loader）
扩展加载器（ExtClassLoader）
系统加载器（AppClassLoader）
用户自定义加载器

在处理字节码相同的时候，并没有自己判断，而且把这个工作返回给上一层，由上一层判断，上一层会继续返回上一层，如果到了跟加载器还是没有发现这个类名（字节码）的话就判定为不存在这个类名（字节码）。

如何破坏双亲委派的话，我们可以举一个例子，比如我创建一个类名为String的类。你看看能不能使用嘛。就算能使用，那你看看他运行后会不会报错嘛。

这个例子就是完全的诠释了双亲委派原则。

1.4 JDK源码除了数据结构你还阅读过哪些？

比如一开始常用的集合包，JUC包，线程等

JUC （java.util.concurrent）

ReentrantLock
ThreadPoolExecutor
AQS

这里介绍一下ReentrantLock。首先介绍一下AQS。AQS的全称是AbstractQueuedSynchronizer 。因为AQS是JUC包下的基类。是比较核心的一个类。

CountDownLatch，ReentrantLock，信号量等都依靠AQS来实现。他可以实现公平锁，也可以实现非公平锁，也是互斥锁以及是可重入锁。

这是ReentrantLock源码的变量 以及 AQS的变量

// 排它锁标识
static final Node EXCLUSIVE = null;
// 如果带有这个标识，证明是失效了
static final int CANCELLED =  1;
// 具有这个标识，说明后继节点需要被唤醒
static final int SIGNAL    = -1;
// Node对象存储标识的地方
volatile int waitStatus;
// 指向上一个节点
volatile Node prev;
// 指向下一个节点
volatile Node next;
// 当前Node绑定的线程
volatile Thread thread;
// 返回前驱节点，如果前期节点为null  就抛出空节点异常
final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
}

// 加锁入口
public void lock() {
    // sync 分为公平和非公平锁
    sync.lock();
}
// 非公平锁
final void lock() {
    // 通过CAS的方式 尝试把state从 0 修改 成1  如果返回true 代表成功，如果返回false  代表失败
    // CAS 是轻量级锁的实现方式
    if (compareAndSetState(0, 1))
        // 将一个属性设置为当前线程，这个属性是AQS的父类提供的  （AbstractOwnableSynchronizer）
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}
private transient Thread exclusiveOwnerThread;
protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
}
// 核心函数
public final void acquire(int arg) {
    //tryAcquire 再次尝试获取锁资源，如果尝试成功 返回true
        if (!tryAcquire(arg) &&
            // 获取锁资源失败后，需要将当前线程封装成一个node 追加到AQS的队列中
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}
// tryAcquire 函数进入后 
final boolean nonfairTryAcquire(int acquires) {
    // 获取当前线程
            final Thread current = Thread.currentThread();
    // 获取AQS的state的值
            int c = getState();
    // 如果 占有锁资源的人 已经释放了 我们就可以尝试获取锁资源
            if (c == 0) {
    // CAS尝试修改state 从0 修改成 1  如果成功 设置 exclusiveOwnerThread 属性为当前线程
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
    // 当前占有锁资源的线程是否是当前线程
            else if (current == getExclusiveOwnerThread()) {
                // 将state +1
                int nextc = c + acquires;
                // 如果加1 小于0 
                if (nextc < 0) // overflow
                    // 抛出超时锁可重入的最大值
                    throw new Error("Maximum lock count exceeded");
                // 如果没问题 继续修改 0 -1 
                setState(nextc);
                // 返回成功
                return true;
            }
            return false;
}
// 从 acquire 函数的 addWaiter
// 说明前面获取锁资源失败，放到锁队列中等待
private Node addWaiter(Node mode) {
    // 创建node类， 并且设置thread 为当前线程，设置为排他锁
        Node node = new Node(Thread.currentThread(), mode);
        // 获取AQS中队列的尾部节点
        Node pred = tail;
     //如果tail 不为null  说明现在队列有人排队
        if (pred != null) {
            // 把当前节点的prev 设置为刚才的尾部节点
            node.prev = pred;
            // 基于CAS的方式 将tail 节点设置为当前节点
            if (compareAndSetTail(pred, node)) {
                // 将之前的节点为next 设置为当前节点
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
}
// 如果当前没有人排队等待 就进入了enq函数  如果CAS失败，也会进入到这个位置重新往队列的尾部插入
private Node enq(final Node node) {
    // 死循环
        for (;;) {
            // 重新获取当前的tail节点为 t
            Node t = tail;
            if (t == null) { 
                // 如果当前tail节点为null  说明现在没人排队，我是第一个
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                // 如果现在有人排队，那我就把尾部插入，继续排队等待
                node.prev = t;
                // 基于CAS的方式 将tail节点设置为当前节点
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
}
// 已经将node 加入到双向队列中，然后执行当前方法
final boolean acquireQueued(final Node node, int arg) {
    // 标识
        boolean failed = true;
        try {
            // 标识
            boolean interrupted = false;
            for (;;) {
                // 获取当前节点的 上一个节点 
                final Node p = node.predecessor();
                // 如果 p是头，说明就是null  那我就执行 利用 tryAcquire函数 再次获取锁资源
                // 获取成功之后 state从 0改成1 返回true 
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // 保证上一个节点是 -1  才会返回true，才会将线程阻塞，等待唤醒获取锁资源
                if (shouldParkAfterFailedAcquire(p, node) &&
                    // 基于Unsafe类的 park方法，挂起线程
                    parkAndCheckInterrupt())  // 针对fail属性，这里是唯一可能出现异常的地方，JVM             内部出现问题时。
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
}
// 如果没有拿到锁资源 执行这个函数 node是当前节点，pred 是上一个节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取上一个节点的状态 state
        int ws = pred.waitStatus;
    // 如果上一个节点状态为 SIGNAL 一切正常
        if (ws == Node.SIGNAL)
            return true;
    // 如果上一个节点已经失效了  （这里失效了我解释一下）
        if (ws > 0) {
            do {
               // 将当前节点的prev指针指向了上一个的上一个
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);  // 一致找到小于等于0的
            // 将重新标识好的最近的有效节点的next
            pred.next = node;
        } else {
            // 小于等于0 ，不等于-1，将上一个有效节点状态修改为 -1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
}
// 上文中 finally 代码块的 cancelAcquire 函数
// 如果当前为null 结束，健壮性判断
private void cancelAcquire(Node node) {
        if (node == null)
            return;

     // 当node的线程置为 null，竞争锁资源跟我没关系了。
        node.thread = null;

        // node的上一个节点
        Node pred = node.prev;
     // node上一个节点 大于0
        while (pred.waitStatus > 0)
            // 找到前面节点中 非失效的节点
            node.prev = pred = pred.prev;
     // 将第一个不是失效节点的下一个节点 声明出来
        Node predNext = pred.next;
  // 当前节点置为失效节点 给别的node看的
        node.waitStatus = Node.CANCELLED;
  // 如果当前节点是尾节点，将尾结点设置为最近的有效节点（如果当前是尾节点）
        if (node == tail && compareAndSetTail(node, pred)) {
            // 用CAS方式将尾节点的next 设置成null
            compareAndSetNext(pred, predNext, null);
        } else {
            int ws;
            // 中间节点操作
            // 如果上一个节点不是头节点
            if (pred != head &&
                // 获取上一节点状态，是不是有效
                ((ws = pred.waitStatus) == Node.SIGNAL || //
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) {
                Node next = node.next;
                if (next != null && next.waitStatus <= 0)
                    compareAndSetNext(pred, predNext, next);
            } else {
                unparkSuccessor(node);
            }

            node.next = node; // help GC
        }
}

1.5 说一下jvm的内存模型？

聊到JVM内存模式，首先聊一下java文件的加载流程吧

通过javac 编译 .java 文件
通过 java 编译成 .class 文件
通过类装载子系统加载到运行时数据区（内存模型） 如下图

堆: 存放的是局部变量信息

栈: 存放的是声明的对象信息，

本地方法栈: 本地方法栈可以通过Thread来查看，start的底层是没有实现的，他的前面加了一个 native。他是通过C，C++实现的。

方法区: 方法区存放的是 .class文件 ，常量，静态变量，类信息。有一些静态变量他也可能说是一种对象如下代码

public static User user = new User();

user会存放在方法区中，但是User会存放在堆上， 在底部会放上最终的流程图，没搞明白的可以去查看一下。

程序计数器: 每个线程都有一个程序计数器。通过字节码执行引擎在执行.class文件的同时调用程序计数器计数。

字节码执行引擎: 用于执行加载进来的.class文件。

我们就举一个main线程的例子。在main函数中执行，整个线程主要包含

程序计数器
栈
栈帧

栈帧又分成 子函数栈帧（类似于调一个自定义函数） 和main函数栈帧

在每一块栈帧中，又包含局部变量，操作数栈，动态链接，方法出口 如下图

局部变量就是我们程序中定义的 int，变量等。
操作数栈就是我们在计算一些信息时，存放的区域。
方法出口就是当调用一个子函数时，有返回信息时，就把返回信息存入方法出口中。从compute栈帧退出到main栈帧后继续执行主函数。

从方法出口出来之后，会到main函数栈帧中，main函数栈帧也是有同样的局部变量，这个局部变量存放的一些对象信息，存的其实不是对象，他存的只是对象在堆中的引用地址

1.6 List、HashMap和HashSet的使用区别？

ArrayList优缺点

有序
可重复
当链表过大时，查询速度慢时间复杂度为O（n）
底层做插入，删除操作时，涉及到copy数据，所以当链表过大，插入删除的性能也很慢

HashMap优缺点

超级快速的查询速度
动态的可变长存储数据（和数组相比较而言）
需要额外计算一次hash值，如果处理不当会占用额外的空间

HashSet优缺点

HashSet基于HashMap实现，以HashSet的值作为HashMap的一个key, 以一个Object对象常量作为HashMap的值。
不允许重复，无序。一般用HashSet要重写hashCode和equals方法

1.7 gcroot都有哪些？老年代和新生代的区别？

gcroots也就是垃圾收集器的对象，gc会收集那些不是gc roots且没有被gc roots引用的对象。一个对象可以属于多个root。

class： 由系统类加载器加载的对象，这些类是不能够被回收的，他们可以以静态字段的方式保存持有其它对象。我们需要注意的一点就是，通过用户自定义的类加载器加载的类，除非相应的java.lang.Class实例以其它的某种（或多种）方式成为roots，否则它们并不是roots，.
Thread： 活着的线程
Stack Local： Java方法的local变量或参数
JNI Local： JNI方法的local变量或参数
JNI Global： 全局JNI引用
Monitor Used: 用于同步的监控对象

将GC Roots 对象作为起点，从这些节点开始向下搜索引用的对象，找到的对象都标记为非垃圾对象，其余未标记的对象都是垃圾对象。

GC Roots根节点：线程栈的本地变量，静态变量，本地方法栈的变量等等。类似于一个树结构

老年代与新生代的区别

存活寿命的不同，老年代的存活寿命大于新生代的存活寿命
新生代采用的是复制算法，老年代采用的是标记删除算法
新生代每次使用的空间不超过90%，主要存放新生的对象。老年代主要存放的是声明周期长的对象。
新生代分为Eden，ServivorFrom，ServivorTo三个区。老年代没有具体的细分区域

Eden：Java新对象的出生地，如果新创建的对象过大会直接分配到老年代。当Eden区内存不够时，就会触发一次MinorGC，对新生代进行一次垃圾回收

ServivorTo：保存Eden区经过一次MintorGC后的幸存者。

ServivorFrom：上一次MintorGC的幸存者，作为这一次的被扫描者。

MintorGC采用的是复制算法，把Eden区，ServivorFrom区域中存活的对象复制到ServivorTo，并且把他们的年代 +1。（如果ServivorTo位置不够用了就会转移部分到老年代，或者年代大于等于15也会转移到老年代）。

然后清空Eden与ServivorFrom的对象，ServivorTo与ServivorFrom互换，原ServivorTo成为下一次GC时的ServivorFrom区了。

老年代的对象比较稳定，MajorGC不会频繁执行，在执行MajorGC前一般都会先执行MintorGC，这样做的好处是使新生代的对象晋升到老年代，导致空间不够用时，再触发。当无法找到足够大的连续空间分配给新创建的较大对象时也会提前触发一次MajorGC进行垃圾回收腾出空间。

因为老年代的对象比较稳定，所以不会采用复制算法，MajorGC会采用标记删除算法，首先扫描一次所有老年代，标记出存活的对象，然后回收没有标记的对象。Major的耗时比较长，因为要扫描再回收。MajorGC会产生内存碎片，为了减少内存损耗，我们一般需要进行合并或者标记出来方便下次直接分配。当老年代也满了装不下的时候，就会抛出OOM异常。

1.8 Heap是什么？和Stack的区别

Heap他是一个堆，Stack是一个栈。

栈的空间是由操作系统自动分配和释放的，堆的空间是手动申请和释放的，也就是我们程序中的new关键字。所以JVM调优绝大多数调优的区域都是在堆中。
存储方式的不同。因为栈的空间是有限的，堆的空间是有很大的自由区，所以我们在使用时，创建的对象，存在栈中往往的引用堆的内存地址，存在堆中才是实际的对象信息。
堆空间用完后，一定要free掉，以防堆溢出，栈就不存在这种问题（如果栈要有这个问题的话，我估计出栈入栈就已经卡住了）

1.9 创建一个对象的过程是什么？

public class Leo {
    public static void main(String[] args) throws CloneNotSupportedException {
        // 通过new关键字创建对象
        VipInfo vipInfo = new VipInfo();
        // 通过clone方式创建对象
        VipInfo vipInfo1 = (VipInfo)vipInfo.clone();
        // 通过 .class.newInstance() 创建对象
        try {
            VipInfo info = VipInfo.class.newInstance();
        }catch (Exception e) {
        }
    }
}

创建对象的方式大概有四种，我的话只了解三种。

通过new
通过clone方式
通过.class方式（类似反射的一种，这里不论述了会在下面的反射那个板块里讲解）

创建流程

当对象被创建时，虚拟机会分配内存来存放对象。这里声明一下从父类，超类继承过来的实例变量也会被分配内存空间。在为这些实例变量分配内存的同时，这些实例变量也会被赋予默认值（零值）。

在内存分配完成之后，Java虚拟机就会开始对新创建的对象按照程序员的意志进行初始化。（这个位置才真正的赋构造的值）在Java对象初始化过程中，分别是 实例变量初始化，实例代码块初始化 和 构造函数初始化

按照程序员的意志进行初始化：这里也可以说是 “半初始化”

1.10 对eaquels的理解？

聊到eaquels函数不得不把 == 聊一下

**==**：判断两个对象的地址是不是相等。即，判断两个对象是不是同一个对象(基本数据类型比较的是值，引用数据类型比较的是内存地址)。

eaquels：

类 没有重写 equals() 方法。则通过 equals() 比较该类的两个对象时，等价于通过“==”比较这两个对象。
类 重写了 equals() 方法。一般，我们都重写 equals() 方法来比较两个对象的内容是否相等；若它们的内容相等，则返回 true (即，认为这两个对象相等)

1.11 HashMap和ConcurrentHashMap的区别？

源码分析的话在数据结构那里已经分析过了，这里简单总结一下区别

HashMap线程是不安全的，ConcurrentHashMap是基于CAS的，所以线程是安全的。
HashMap允许Key与Value为空，ConcurrentHashMap不允许
HashMap不允许通过迭代器遍历的同时修改，ConcurrentHashMap允许。并且更新可见
他们俩的底层数据结构都是一样的（数组+链表+红黑树）

1.12 HashMap为什么用红黑树？

不是用红黑树来管理HashMap的，而是在hash相同的情况下，且重复数量大于8。用红黑树来管理数据。红黑树相当于排序数据，可以自动的使用二分法进行定位，性能较高。一般情况下，hash值做的比较好的话基本上用不到红黑树。

红黑树牺牲了一些查找性能但其本身并不是完全平衡的二叉树。因此插入删除操作效率略高于AVL树。 AVL树用于自平衡的计算牺牲了插入删除性能，但是因为最多只有一层的高度差，查询效率会高一些。

1.13 JVM调优参数有哪些？随便说几个

-Xms256m：初始化堆大小为 256m；
-Xmx2g：堆最大内存为 2g；
-Xmn50m：新生代的大小50m；
-XX:+PrintGCDetails 打印 gc 详细信息
-XX:+HeapDumpOnOutOfMemoryError 在发生OutOfMemoryError错误时，来dump堆快照
-XX:NewRatio=4 设置年轻的和老年代的内存比例为 1:4；
-XX:SurvivorRatio=8 设置新生代 Eden 和 Survivor 比例为 8:2；
-XX:+UseSerialGC 新生代和老年代都用串行收集器 Serial + Serial Old
-XX:+UseParNewGC 指定使用 ParNew + Serial Old 垃圾回收器组合；
-XX:+UseParallelGC 新生代使用Parallel Scavenge，老年代使用Serial Old
-XX:+UseParallelOldGC：新生代ParallelScavenge + 老年代ParallelOld组合；
-XX:+UseConcMarkSweepGC：新生代使用ParNew，老年代的用CMS；
-XX:NewSize；新生代最小值；
-XX:MaxNewSize：新生代最大值
-XX:MetaspaceSize 元空间初始化大小
-XX:MaxMetaspaceSize 元空间最大值

1.14 反射是什么？

反射是把Java类中的各个成分映射成一个个的Java对象。在运行状态中

对于任意一个类，都能够知道这个类的所有属性和方法
对于任意一个对象，都能够调用它的任意一个属性和方法

优点

在程序运行过程中可以操作类对象，增加了程序的灵活性；
解耦，从而提高程序的可扩展性，提高代码的复用率，方便外部调用；
对于任何一个类，当知道它的类名后，就能够知道这个类的所有属性和方法；而对于任何一个对象，都能够调用它的一个任意方法。

缺点

性能问题：Java 反射中包含了一些动态类型，JVM 无法对这些动态代码进行优化，因此通过反射来操作的方式要比正常操作效率更低。
安全问题：使用反射时要求程序必须在一个没有安全限制的环境中运行，如果程序有安全限制，就不能使用反射。
程序健壮性：反射允许代码执行一些平常不被允许的操作，破坏了程序结构的抽象性，导致平台发生变化时抽象的逻辑结构无法被识别。

1.15 Java中创建多线程的方式有哪些？线程池有哪些参数？

创建方式

继承Thread类，重写run方法
实现Runnable接口，重写run方法
通过Callable和FutureTask创建线程
通过线程池创建线程

常用参数如下

线程池的线程个数 nThreads
创建新线程时要使用的工厂 threadFactory
线程池中的核心线程数量，即使这些线程没有任务干，也不会将其销毁 corePoolSize
当线程池中的线程数量大于corePoolSize时，多余的线程等待新任务的最长时间 keepAliveTime
keepAliveTime的时间单位。 unit
在线程池中的线程还没有还得及执行任务之前，保存任务的队列 workQueue
当线程池中没有多余的线程来执行任务，并且保存任务的多列也满了（指的是有界队列），对仍在提交给线程池的任务的处理策略 handler RejectedExecutionHandler

线程池的好处

降低资源消耗：可以重复利用已创建的线程降低线程创建和销毁造成的消耗。
提高响应速度：当任务到达时，任务可以不需要等到线程创建就能立即执行。
提高线程的可管理性：线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控

1.16 执行流程和拒绝策略是什么？

执行流程

当有新任务提交到线程池时，当线程数<核心线程数，创建线程
如果当前线程数>=核心线程数，且任务队列未满时，将任务放入任务队列
如果当前线程数>=核心线程数，且任务队列已满时，根据拒绝策略进行相应的处理
如果当前线程数<最大线程数，创建线程
如果当前线程数=最大线程数，调用拒绝执行处理程序

源码分析

/**
 * 在将来的某个时间执行给定的任务。 这个任务可以在新线程中执行，也可以在现有的线程池中执行。  
 * 如果任务不能提交执行，可能是因为这个原因“执行者”已经被关闭，或者因为它的容量已经达到，  
 * 当前任务就由RejectedExecutionHandler 进行处理
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * 主要分三步走
     * 1. 如果运行的线程少于corePoolSize，则尝试用给定的命令作为第一个命令启动一个新的线程
     * 任务。 调用addWorker会自动检查runState和workerCount，这样可以防止误报 
     * 线程，当它不应该返回false。
     *
     * 2. 如果一个任务可以成功排队，那么我们仍然需要再次检查我们是否应该添加一个线程  
     * (因为现有的已经在上次检查后死亡)或其他进入此方法后池关闭。 所以我们重新检查状态，
     * 如果有必要，回滚排队停止，或启动一个新的线程(如果没有)。  
     *
     * 3. 如果不能对任务进行排队，则尝试添加一个新的线程。 如果它失败了，我们知道我们已经关闭或饱和了      * 因此拒绝任务。  
     */
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

拒绝策略

很清晰的看到，当前共有四中实现类实现这个接口。

AbortPolicy：该策略是直接将提交的任务抛弃掉，并抛出RejectedExecutionException异常。

/**
 * A handler for rejected tasks that throws a
 * {@code RejectedExecutionException}.
 */
public static class AbortPolicy implements RejectedExecutionHandler {
    /**
     * Creates an {@code AbortPolicy}.
     */
    public AbortPolicy() { }

    /**
     * Always throws RejectedExecutionException.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     * @throws RejectedExecutionException always
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

DiscardPolicy:该策略也是将任务抛弃掉（对于提交的任务不管不问，什么也不做），不过并不抛出异常。

/**
 * A handler for rejected tasks that silently discards the
 * rejected task.
 */
public static class DiscardPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code DiscardPolicy}.
     */
    public DiscardPolicy() { }

    /**
     * Does nothing, which has the effect of discarding task r.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

DiscardOldestPolicy : 该策略是当执行器未关闭时，从任务队列workQueue中取出第一个任务，并抛弃这第一个任务，进而有空间存储刚刚提交的任务。使用该策略要特别小心，因为它会直接抛弃之前的任务。

/**
 * A handler for rejected tasks that discards the oldest unhandled
 * request and then retries {@code execute}, unless the executor
 * is shut down, in which case the task is discarded.
 */
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code DiscardOldestPolicy} for the given executor.
     */
    public DiscardOldestPolicy() { }

    /**
     * Obtains and ignores the next task that the executor
     * would otherwise execute, if one is immediately available,
     * and then retries execution of task r, unless the executor
     * is shut down, in which case task r is instead discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

CallerRunsPolicy:该策略并没有抛弃任何的任务，由于线程池中已经没有了多余的线程来分配该任务，该策略是在当前线程（调用者线程）中直接执行该任务。

/**
 * A handler for rejected tasks that runs the rejected task
 * directly in the calling thread of the {@code execute} method,
 * unless the executor has been shut down, in which case the task
 * is discarded.
 */
public static class CallerRunsPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code CallerRunsPolicy}.
     */
    public CallerRunsPolicy() { }

    /**
     * Executes task r in the caller's thread, unless the executor
     * has been shut down, in which case the task is discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

1.17 Java的序列化和反序列化？

Java序列化指将Java对象转换为字节序列的过程，反序列化指将字节序列转换为目标对象的过程；

当Java对象需要网络传输或者持久化到磁盘上时，才需要进行序列化。

二、MySQL

MySQL

2.1 MySQL的事务和引擎等问题

事务是一种机制，一个操作序列，包含一组数据库操作命令并且把所有命令作为一个整体统一向系统提交或撤销操作请求，即这一组数据库命令要么都执行，要么都不执行。

事务的特点的话就是常说的 ACID

A（原子性）：要不成功，要不失败
C（一致性）：在事务开始之前和事务结束以后，数据库的完整性约束没有被破坏。
I（隔离性）：在并发环境中，当不同的事务同时操纵相同的数据时，每个事务都有各自的完整数据空间
D（持久性）：在事务完成以后，该事务所对数据库所作的更改便持久的保存在数据库之中，并不会被回滚。

事务的隔离级别

Read uncommitted（读未提交）：事务中的修改即使没有提交对其他事务都是可见的，也可以称为脏读，这个级别会导致很多问题，从性能上来说不会比其他隔离级别好太多，但缺乏其他隔离级别的很多好处。除非真的有特定的需求，一般很少用
Read committed（读提交）：大多数数据库默认的都是read committed，但是MySQL默认的不是这个！一个事务从执行到提交前，其他事务都是不可见的，有时候也可以叫不可重复读，因为两次执行同样的查询可能会得到不一样的查询结果
Repeatable read（可重复读）：repeatable read解决了read committed脏读的问题，这个隔离级别也是MySQL默认的隔离级别。该级别保证了同一个事务多次执行可以读取同样的数据，但是有个缺陷就是存在幻读！幻读就是当事务在某个范围内读取数据时，这时另一个事务在这个范围插入了数据，当读取的事务再次读取该范围时会产生幻行。通过多版本并发控制（MVCC）解决了幻读的问题。
Serializable（串行化）：这是最高的隔离级别，它通过强制事务在从串行上执行，避免了前面说的幻读问题，简单来说就在在读取数据时加一个锁，这就暴露了另一个问题，大量的加锁会导致出现争锁超时的问题。只有特定的需求情况下或者可以接收没有并发的情况下才考虑这种隔离级别。

MySQL常用的引擎主要有两种 MyISAM 和 InnoDB

MyISAM

MyISAM不支持事务，也不支持外键约束，只支持全文索引，数据文件和索引文件是分开保存的。访问速度快，对事务完整性没有要求。MyISAM适合查询、插入为主的应用场景

MyISAM在磁盘上存储成三个文件，文件名和表名都相同，但是扩展名分别为:

文件存储表结构的定义.frm
数据文件的扩展名为.MYD (MYData)
索引文件的扩展名是.MYI(MYIndex)

表级锁定形式,数据在更新时锁定整个表数据库在读写过程中相互阻塞（串行操作，按照顺序操作，每次在读或写的时候会把全表锁起来）会在数据写入的过程阻塞用户数据的读取也会在数据读取的过程中阻塞用户的数据写入特性：数据单独写入或读取,速度过程较快且占用资源相对少

MyISAM表支持三种不同的存储格式

静态（固定长度)表： 静态表是默认的存储格式。静态表中的字段都是非可变字段，这样每个记录都是固定长度的，这种存储方式的优点是存储非常迅速，容易缓存，出现故障容易恢复;缺点是占用的空间通常比动态表多。
动态表： 动态表包含可变字段(varchar)，记录不是固定长度的，这样存储的优点是占用空间较少，但是频繁的更新、删除记录会产生碎片，需要定期执行OPTIMIZE TABLE 语句或myisamchk -r 命令来改善性能，并且出现故障的时候恢复相对比较困难。
压缩表： 压缩表由 myisamchk工具创建，占据非常小的空间，因为每条记录都是被单独压缩的，所以只有非常小的访问开支

MyISAM适应场景

公司业务不需要事务的支持
单方面读取或写入数据比较多的业务
MyISAM存储引擎数据读写都比较频繁场景不适合
使用读写并发访问相对较低的业务
数据修改相对较少的业务
对数据业务一致性要求不是非常高的业务服务器硬件资源相对比较差

InnoDB

InnoDB支持事务，外键，InnoDB不支持FULLTEXT类型的索引

InnoDB适合频繁修改以及涉及到安全性较高的应用。

InnoDB中不保存表的行数，如select count(*) from table时，InnoDB需要扫描一遍整个表来计算有多少行

清空整个表时，InnoDB是一行一行的删除，效率非常慢。MyISAM则会重建表

InnoDB支持行锁（某些情况下还是锁整表，如 update table set a=1 where user like ‘%lee%’

在磁盘存储：基于磁盘的资源是InnoDB表空间数据文件和它的日志文件，InnoDB 表的大小只受限于操作系统文件的大小，一般为 2GB

2.2 MySQL的调优看过吗？有调优经验吗？

当查询一条数据时，加上 limit 1
为搜素字段，经常查询的字段加索引（一定不要给区分度不高的字段加索引，比如订单状态）
作关联查询时，一定要保证两个字段类型相同，索引相同。
查询肯定是避免 select *
字段值一定要设置 not null，减少存储空间而且进行比较时程序会更复杂。
设计表时，考虑固定长度的表会更快（表中没有如下类型的字段： VARCHAR，TEXT，BLOB）
越小的列越快（根据字段长度的大小设置对应的字段长度）
in ， not in，!=，<> 要慎用，否则会导致全表扫描
对于大表来说，使用偏移量查询数据可以采用子查询limit 嵌套大查询limit方式
谨慎使用order by 字段
禁用Query Cache
如果单表数据量较大的话，则进行分表，分表的单表数据量应该维持在1000万以内。
如果单库出现瓶颈，则进行分库，读写分离。
如果物理机器性能瓶颈，可以增加多个数据库节点，搭建分布式集群。主从库等等。
根据需求进行合理的存储。（比如身份证信息，如果要取后六位信息进行根据地区统计的话，如果按照正常存储，那就要建立18位的索引，索引越大占用空间就越大，数据页放的索引就越少，搜索效率就会越低。所以如果我们倒叙存储的话，我们建立索引时就可以直接建立一个6个字节。）
这里可以从innodb_buffer_pool_size，innodb_flush_log_at_trx_commit ，innodb_io_capactiy 这类参数入手，奈何我技术实力还不够，后续会慢慢深入

2.3 MySQL的索引失效有哪些场景？

like
计算函数、函数处理
类型不匹配，隐式类型转换
or
<>
OR语句前后没有同时使用索引
联合索引ABC问题

2.4 你在上文提到了慎用order by，可以聊聊原理吗？

-- 表定义
CREATE TABLE `t` (
  `id` int(11) NOT NULL,
  `city` varchar(16) NOT NULL,
  `name` varchar(16) NOT NULL,
  `age` int(11) NOT NULL,
  `addr` varchar(128) DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `city` (`city`)  -- 为了避免全表扫描，我们给city加了索引，这样先去找 `city = '常州'` 的时						 候就不需要全表了。
) ENGINE=InnoDB;
-- 查询语句
select city,name,age from t where city='常州' order by name limit 1000;

首先order by 有两种排序算法 全字段排序 和 rowid 排序

全字段排序

因为是order by语句，一开始会初始化一个sort_buffer 然后放入city,name,age 这三个字段
先从索引树上找到第一个符合city= '常州' 的数据的id （city，id都是索引，所以通过city是可以拿到当前 city=常州 对应的id）
去主键id索引取出整行数据，也就是select用到的city,name,age 存入sort_buffer
再回到第二步，第三步重复操作，直至完成所有数据。
查完之后对 sort_buffer 的数据按照字段name快速排序
按照排序后的结果取出前1000行的数据

第五步中，有一个按照字段name快速排序，这里是有一个参数控制的。

sort_buffer_size：MySQL 为排序开辟的内存（sort_buffer）的大小。如果要排序的数据量小于 sort_buffer_size，排序就在内存中完成。但如果排序数据量太大，内存放不下，则不得不利用磁盘临时文件辅助排序。

外部： 当借助外部排序时，一般使用的是 归并排序算法。他会把整个sort_buffer分成多个临时文件，每一份临时文件单独进行排序，然后再把12个有序文件再合并成一个有序的大文件。

如果 sort_buffer_size 超过了需要排序的数据量的大小，number_of_tmp_files 就是 0，表示排序可以直接在内存中完成。

否则就需要放在临时文件中排序。sort_buffer_size 越小，需要分成的份数越多，number_of_tmp_files 的值就越大。

rowid排序

上面算法中，采用的是把数据统一写入sort_buffer或者临时中，但是这个算法有个问题，如果返回的字段很多的话，sort_buffer里放的字段数太多，导致内存中的行数太少。要分成很多个临时文件，排序的性能就会很差。

除了全字段排序，MySQL还引用了另一种排序算法， rowid排序

rowid排序需要先设置一个参数 SET max_length_for_sort_data = 16;

max_length_for_sort_data，是 MySQL 中专门控制用于排序的行数据的长度的一个参数。它的意思是，如果单行的长度超过这个值，MySQL 就认为单行太大，要换一个算法

根据city，name，age 的字段长度计算得知是 16+16+11（4） = 36，我们把 max_length_for_sort_data 设置为16。新的算法放入 sort_buffer 的字段，只有要排序的列（即 name 字段）和主键 id。但这时，排序的结果就因为少了 city 和 age 字段的值，不能直接返回了，整个执行流程就变成如下所示的样子：

初始化 sort_buffer ，确定放入两个字段，name 和 id
从索引city上找到第一个满足杭州的数据，也就是主键id
再去主键索引上取出整行，也就是 name，id 存入sort_buffer
重复第二步和第三步直至不满足条件为止。
为 sort_buffer 按照name排序
遍历排序结果，取前1000行，并按照id的值回到原表取出这个id 对应的city,name,age三个字段返回给客户端

rowid比全字段排序多了最后一步回表的过程。实际上MySQL服务器端从排序后的sort_buffer取出id，然后去原表查city，name，age字段时，不需要在服务端再耗费内存存储结果，是直接返回给客户端的。

从number_of_tmp_files结果可以得知。

number_of_tmp_files变小了，扫描的数据是不变的，但是字每一行变小了，因此需要排序的总数据量就变小了，需要的临时文件也相应地变少了。

排序算法分析

两种算法各有各的好处吧，实际的实战中我们应该如何选择呢？

如果 MySQL 实在是担心排序内存太小，会影响排序效率，才会采用 rowid 排序算法，这样排序过程中一次可以排序更多行，但是需要再回到原表去取数据。
如果 MySQL 认为内存足够大，会优先选择全字段排序，把需要的字段都放到 sort_buffer 中，这样排序后就会直接从内存里面返回查询结果了，不用再回到原表去取数据。

这也体现了MySQL的一个设计思想：如果内存够，就多用内存，减少磁盘的访问。

其实我们可以换一个思路解决，如果输出的字段不多，我们可以通过建立联合索引 + 插入时就按照那个字段排序 这样就不需要使用order by，sort_buffer了。

2.5 基本语句，如何增加列？

在一个表加一个列的话，首先应该看是大表还是小表。如果是小表的话，我感觉随意操作。大表的话就要留意一下加字段的时候会涉及到锁表操作。一旦锁表就会暂停这个表的所有业务。

我们可以通过创建一个中间表，然后转移那个表的索引信息。

最后通过insert into user (X,Y,Z) select X,Y,Z from user 转移真实数据。

最好在脱机的情况下执行，以免在迁移数据的时候，有新数据写入，导致新表数据流失不完整。

2.6 数据库主从复制原理

什么是主从复制？

主从复制是指一个MySQL数据库服务器的主节点复制到一个或多个从节点。默认采用异步复制方式，这样从节点不用一直访问主服务器来更新自己的数据，数据的更新可以在远程连接上进行，从节点可以复制主数据库中的所有数据库或特定的数据库，特定的数据表。

为什么需要主从复制？

防止在做写操作时，发生锁表，导致读请求无法响应。
数据备份，一旦主库挂了或者被删了，也可以通过从库做数据恢复
数据量变大之后，IO访问率过高，单机无法满足。通过多台机器降低磁盘IO访问的评率，提升单台机器的IO性能。

主从复制原理？

master服务器将数据的改变记录记录二进制binlog日志，当maser上的数据发生改变时，则将其写入二进制日志文件中。
slave服务器会在一定时间间隔内对master二进制日志进行探测其是否发生改变，如果发生改变，则开始一个IOThread请求master二进制事件
同时主节点为每个IO线程启动一个dump线程，用于向其发送二进制事件，并保存至从节点本地的relay log中，从节点将启动SQL线程从relay log中读取二进制日志，在本地重放，使得其数据和主节点的保持一致，最后IOThread和SQLThread将进入睡眠状态，等待下一次被唤醒。

如何保证主从数据一致？

在主从复制原理的基础上可以再深聊一下，主从数据是如何做到一致性的！主从数据同步主要是通过binlog日志

从库上通过change master 命令，设置主库的IP，端口，用户名，密码以及日志文件名和偏移量。
在从库上执行 start slave 命令，这个时候从库会启动两个线程。一个是 io_thread 和 sql_thread 其中 io_thread 主要负责与主库建立连接。
主库校验完用户名，密码，开始按照从库传过来的位置，从本地读取binlog发送给从库。
从库拿到主库传过来的binlog后，写到本地文件，这个本地文件就是常说中继文件（relay log）
sql_thread 读取中转日志，解析出日志里的命令，并执行。这样就达到了主从数据一致性的要求。

binlog是啥

上面聊了很多binlog，这里我们细聊一下binlog是啥。面试的时候肯定会细问的，毕竟是阿里面试，不可能问的那么粗浅。

binlog他是mysql的二进制日志，它记录了所有的DDL和DML语句。它是以事件形式记录的，还会包含所执行的消耗时间。binlog的主要目的是复制和恢复

binlog主要有三种格式 row ，statement ，mixed

binlog_format = statement

begin
在执行真实的SQL语句之前，会有一个 use 数据库名 命令。这条命令不是我们主动执行的，而是 MySQL 根据当前要操作的表所在的数据库，自行添加的。这样做可以保证日志传到备库去执行的时候，不论当前的工作线程在哪个库里，都能够正确地更新到 test 库的表 t
到了中间部分就是完整的记录一个SQL的所有信息，连注释都会一起记录。
最后部分是COMMIT 。提交

这个参数是设置当前binlog 按照哪种格式存储的 binlog_format

binlog_format = row

begin
有两个 event：Table_map 和 Delete_rows。
commit，提交

Table_map event，用于说明接下来要操作的表是 test 库的表 t;

Delete_rows event，用于定义删除的行为。

通过以上信息是无法定位的，需要借助mysqlbinlog 工具才可以。

row格式的优点就是binlog日志记录是真实删除行的主键id，这样binlog日志传到备库的时候就会肯定删除对应的id。

statement的格式会有一个误操作。比如举个例子

-- a字段是一个索引
delete from user where a = 1 limit 1

如果是上述数据的话，有可能根据不同库的索引不同，导致删除不同的数据，最终导致主从库大量数据不一致。

binlog_format = mixed

之所以有mixed这种格式主要取决于三点。

因为有些 statement 格式的 binlog 可能会导致主备不一致，所以要使用 row 格式。
但 row 格式的缺点是，很占空间。比如你用一个 delete 语句删掉 10 万行数据，用 statement 的话就是一个 SQL 语句被记录到 binlog 中，占用几十个字节的空间。但如果用 row 格式的 binlog，就要把这 10 万条记录都写到 binlog 中。这样做，不仅会占用更大的空间，同时写 binlog 也要耗费 IO 资源，影响执行速度
所以，MySQL 就取了个折中方案，也就是有了 mixed 格式的 binlog。mixed 格式的意思是，MySQL 自己会判断这条 SQL 语句是否可能引起主备不一致，如果有可能，就用 row 格式，否则就用 statement 格式。（比较重要）

2.7 MySQL中都有哪些锁？

行锁与表锁

只有明确指定主键，才会执行行锁，否则执行表锁。

无锁

-- 当前这个表不存在主键
select * from user where id = -1 for update;

行锁

select * from user where id = -1 for update;
select * from user where id = -1 and name = 'kkkk' for update;

表锁

-- 主键不明确
select * from user where name = 'kkk' for update;
select * from user where id <>

锁算法（机制）

行锁算法

Record Lock（普通行锁）

键值在条件范围内
记录存在

GapLock（间隙锁）

对于键值不存在条件范围内，叫做 “间隙” (GAP)，引擎就会对这个间隙加锁，这种机制就是Gap机制

Next-Key Lock（行&间隙锁）

在键值范围条件内，同时键值又不存在条件范围内

表锁算法

意向锁（升级机制）

当一个事务带着表锁去访问一个被加了行锁的资源，那么，此时这个行锁就会升级成意向锁，将表锁住。

-- 事务A  与 事务B  当前id=10这条数据中， name的值为kkk  就会升级
select * from user where id = 10 for update
select * from user where name like 'kkk%' for update

自增锁

事务插入自增类型的列时，获取自增锁

如果一个事务正在往表中插入自增记录，其他事务都必须等待

实现

行锁和表锁其实是粒度的概念，共享锁和排它锁是他们的具体实现

共享锁

允许一个事务去读一行，组织其他事务去获取该行的排它锁
一般理解：能读，不能写

排它锁

允许持有排它锁的事务读写数据，阻止其他事务获取该资源的共享锁和排它锁。
不能获取任何锁，不代表不能读

注意点

某个事务获取数据的排它锁，其他事务不能获取该数据的任何锁，并不代表其他事务不能无锁读取该数据

乐观锁

一般通过版本号进行更新操作

update user set name = 'kkk' where id = 1

悲观锁

每次获取数据时，对该记录加排它锁，期间其他用户阻塞等待访问该记录。悲观锁适合写入频繁的场景。

select * from user where id = 1 for update;
update user set buy_sum = buy_sum - 1 where id =1;

总结

InnoDB行锁是通过给索引上的索引项加锁来实现的，只有通过索引条件检索数据，InnoDB才使用行级锁，否则，InnoDB将使用表锁。
由于MySQL的行锁是针对索引加的锁，不是针对记录加的锁，所以虽然是访问不同行的记录，但是如果是使用相同的索引键，是会出现锁冲突的。应用设计的时候要注意这一点。
当表有多个索引的时候，不同的事务可以使用不同的索引锁定不同的行，另外，不论是使用主键索引、唯一索引或普通索引，InnoDB都会使用行锁来对数据加锁。
即便在条件中使用了索引字段，但是否使用索引来检索数据是由MySQL通过判断不同执行计划的代价来决定的，如果MySQL认为全表扫描效率更高，比如对一些很小的表，它就不会使用索引，这种情况下InnoDB将使用表锁，而不是行锁。因此，在分析锁冲突时，别忘了检查SQL的执行计划，以确认是否真正使用了索引。
检索值的数据类型与索引字段不同，虽然MySQL能够进行数据类型转换，但却不会使用索引，从而导致InnoDB使用表锁。通过用explain检查两条SQL的执行计划，我们可以清楚地看到了这一点。

2.8 InnoDB引擎为什么使用B+树？

索引结构是主要分五块

哈希：只支持单值查询，不支持范围查询，所以innodb引擎无法以哈希结构为主结构
链表：体量增大后，对查询的时间复杂度还是比较鸡肋的，O(n)。
二叉树：使用二叉树时，如果当前插入的值的确是持续递增的就会出现树节点一边倒的情况，这样的话就有点类似于链表了，查询性能不好。
红黑树：在一定程序上的确解决了平衡问题，但是还没有完成解决。出现了层级较多这个问题。层级较多会影响查询性能
B+树：在B树的基础上作了优化，也是红黑树之后的一个进化版。主要优化点就是数据节点的自旋。在插入时，当节点树大于某一个限制后会自动自旋，变成另一个节点树。而且具有排序的功能。节点与节点之间有连接关系，这是对查询非常有利的。

综合每个结构的优缺点进行权衡，最终选择B+树

2.9 如何发现慢SQL

首先要检查是否开启慢查询日志

slow_query_log 默认是off关闭的，使用时，需要改为on 打开

slow_query_log_file 记录的是慢日志的记录文件

long_query_time 默认是10S，每次执行的sql达到这个时长，就会被记录

查看慢查询状态

Slow_queries 记录的是慢查询数量当有一条sql执行一次比较慢时，这个vlue就是1 （记录的是本次会话的慢sql条数）

如果显示非0，那么就是慢查询了，我们就可以根据具体的SQL去进行explain，检查执行计划。进行相应的SQL优化。

不走慢查询的话也可以通过，在编写SQL时进行习惯性的explain调优一下。

慢查询工具的话可以借助 mysqldumpslow 。优化策略可以参考上面的索引，调优那个板块。

三、Spring

Spring

3.1 对Spring的理解？

概念

spring是Java企业级应用的开源开发框架

好处

轻量：Spring是轻量的，基本的版本大约2MB

控制反转：Spring通过控制反转实现了松散耦合，对象们给出它们的依赖，而不是创建或查找依赖的对象们

**面向切面的编程(AOP)**：Spring支持面向切面的编程，并且把应用业务逻辑和系统服务分开

容器：Spring包含并管理应用中对象的生命周期和配置

MVC框架：Spring的WEB框架是个精心设计的框架，是Web框架的一个很好的替代品

事务管理：Spring提供一个持续的事务管理接口，可以扩展到上至本地事务下至全局事务（JTA）

异常处理：Spring提供方便的API把具体技术相关的异常（比如由JDBC，HibernateorJDO抛出的）转化为一致的unchecked异常

3.2 spring 和 springboot的区别？

springboot是spring家族中的一员，我觉得这两个都不在同一个纬度。说句心里话这道面试题我怀疑不是阿里面试真题。

3.3 说一下Bean的生命周期？注入过程？

1、解析xml配置或注解配置的类，得到BeanDefinition;

2、通过BeanDefinition反射创建Bean对象;

3、对Bean对象进行属性填充;

4、回调实现了Aware接口的方法，如BeanNameAware;

5、调用BeanPostProcessor的初始化前方法;

6、调用init初始化方法;

7、调用BeanPostProcessor的初始化后方法，此处会进行AOP;

8、将创建的Bean对象放入一个Map中;

9、业务使用Bean对象;或着是通过洼解配置的

10、Spring容器关闭时调用DisposableBean的destory)方法;

3.5 Spring Boot启动类注解是什么，有什么作用，自动装配原理？

// 目的是开启springboot的自动配置
@SpringBootApplication

由源码我们得知@SpringBootApplication 包含下列七种注解

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@SpringBootConfiguration
@EnableAutoConfiguration
@ComponentScan()
public @interface SpringBootApplication {

}

@Target(ElementType.TYPE) ：说明了Annotation（注解）所修饰的对象范围

1.CONSTRUCTOR:用于描述构造器 2.FIELD:用于描述域 3.LOCAL_VARIABLE:用于描述局部变量 4.METHOD:用于描述方法 5.PACKAGE:用于描述包 6.PARAMETER:用于描述参数 7.TYPE:用于描述类、接口(包括注解类型) 或enum声明

@Retention(RetentionPolicy.RUNTIME)： 注解按生命周期来划分可分为3类：

1、RetentionPolicy.SOURCE：注解只保留在源文件，当Java文件编译成class文件的时候，注解被遗弃； 2、RetentionPolicy.CLASS：注解被保留到class文件，但jvm加载class文件时候被遗弃，这是默认的生命周期； 3、RetentionPolicy.RUNTIME：注解不仅被保存到class文件中，jvm加载class文件之后，仍然存在；

@Documented： 这个注解只是用来标注生成javadoc的时候是否会被记录。

@Inherited： 是一个标识，用来修饰注解，自定义注解当中会用到

@SpringBootConfiguration： 标注在某个类上，表示这是一个Spring Boot的配置类。（这个注解也是一个自定义注解）

@EnableAutoConfiguration： 需要配置的东西，Spring Boot会帮我们自动配置；

@EnableAutoConfiguration告诉SpringBoot开启自动配置功能；这样自动配置才能生效；

点进去会发现@Import，说白了他就是借助@Import的支持，收集和注册特定场景相关的bean定义。

@Import作用：用于导入其他的配置类。而@EnableAutoConfiguration也是借助@Import的帮助，将所有符合自动配置条件的bean定义加载到IoC容器，仅此而已！

EnableAutoConfigurationImportSelector：导入哪些组件的选择器；会给容器中导入非常多的自动配置类（xxxAutoConfiguration）；

大概的流程：

Spring Boot在启动的时候，通过EnableAutoConfigurationImportSelector类，从类路径下的 META-INF/spring.factories中获取EnableAutoConfiguration指定的值（就是上方截图），以全类名反射的创建方式，将这些值作为自动配置类导入到容器中，自动配置类就生效，帮我们进行自动配置工作；

以前我们需要自己配置的东西，自动配置类都帮我们配置好了，这也就是使用springboot在使用spring，springmvc不用配置视图解析器、数据库连接池、事务等配置的原因。直接开箱即用。

当然springboot也给我提供了修改配置的方法，那就是通过yml或者propertie文件来进行修改springboot为我们配置好的配置默认值。

@ComponentScan： 用于通过注解指定spring在创建容器时要扫描的包

我们可以通过basePackages等属性来细粒度的定制@ComponentScan自动扫描的范围，如果不指定，则默认Spring框架实现会从声明@ComponentScan所在类的package进行扫描。

3.6 定时框架用过吗？

我了解的定时框架大概有三种

quartz

调用API的的方式操作任务，不人性化；
调度逻辑和QuartzJobBean耦合在同一个项目中，这将导致一个问题，在调度任务数量逐渐增多，同时调度任务逻辑逐渐加重的情况加，此时调度系统的性能将大大受限于业务；
Quartz关注点在于定时任务而非数据，并无一套根据数据处理而定制化的流程。虽然Quartz可以基于数据库实现作业的高可用，但缺少分布式并行调度的功能。
支持分布式高可用，我们需要某个定时任务在多个节点中只有某个节点可以执行时，就需要Quartz来实现，否则使用@Scheduled等方式会造成所有节点都执行一遍。
支持持久化，Quartz有专门的数据表来实现定时任务的持久化。
支持多任务调度和管理，Quartz可以在数据库中存储多个定时任务进行作业调度，可以实现定时任务的增删改查等管理。

xxl-job

侧重的业务实现的简单和管理的方便，学习成本简单，失败策略和路由策略丰富。推荐使用在“用户基数相对少，服务器数量在一定范围内”的情景下使用。

elastic-job

关注的是数据，增加了弹性扩容和数据分片的思路，以便于更大限度的利用分布式服务器的资源。但是学习成本相对高些，推荐在“数据量庞大，且部署服务器数量较多”时使用。

这里我常用的是 Quartz，Quartz由三部分组成：

任务：JobDetail
触发器：Trigger（分为SimpleTrigger和CronTrigger）
调度器：Scheduler

JobDetail主要由JobKey（job的名字name和分组group）、JobClass、JobDataMap（任务相关的数据）、JobBuilder组成。常用的是前几个。

Trigger规定触发执行Job实现类，主要有SimpleTrigger和CronTrigger两个实现类。Trigger由以下部分组成：

TriggerKey（job的名字name和分组group）

JobDataMap（Trigger相关的数据，同JobDetail中JobDataMap，存相同key，若value不同，会覆盖前者。）

ScheduleBuilder（有CronScheduleBuilder、SimpleScheduleBuilder、CalendarIntervalScheduleBuilder、DailyTimeIntervalScheduleBuilder常用前2种。）

Scheduler 调度器就是为了读取触发器Trigger从而触发定时任务JobDetail。可以通过SchedulerFactory进行创建调度器，分为StdSchedulerFactory（常用）和DirectSchedulerFactory两种。

StdSchedulerFactory使用一组属性（放在配置文件中）创建和初始化调度器，然后通过getScheduler()方法生成调度程序。

DirectSchedulerFactory不常用，容易硬编码。

3.7 Spring IOC和AOP讲讲你的理解？

IOC

IOC(Inversion of Control) ,即控制反转，不是具体的技术，而是一种思想,IOC意味着将你设计好的对象交给Spring容器来管理，而不是传统的在你的对象内部直接控制。对于spring框架来说，就是由spring来负责控制对象的生命周期和对象间的关系。

IOC或DI（Dependency Injection，依赖注入）把应用的代码量降到最低。它使应用容易测试，单元测试不再需要单例和JNDI查找机制。最小的代价和最小的侵入性使松散耦合得以实现。IOC容器支持加载服务时的饿汉式初始化和懒加载。

Spring IOC负责创建对象、管理对象，通过依赖注入（DI），装配对象，配置对象，并且管理这些对象的整个生命周期。

IOC主要有两种注入方式

构造器依赖注入：构造器依赖注入通过容器触发一个类的构造器来实现的，该类有一系列参数，每个参数代表一个对其他类的依赖。
Setter方法注入：Setter方法注入是容器通过调用无参构造器或无参static工厂方法实例化bean之后，调用该bean的setter方法，即实现了基于setter的依赖注入。

AOP

如果说 IoC 是 Spring 的核心，那么面向切面编程就是 Spring 最为重要的功能之一了，在数据库事务中切面编程被广泛使用。

AOP模块用于发给我们的Spring应用做面向切面的开发，这样就确保了Spring和其他AOP框架的共通性。

AOP能够将那些与业务无关，却为业务模块所共同调用的逻辑或责任（例如事务处理、日志管理、权限控制等）封装起来，便于减少系统的重复代码，降低模块间的耦合度，并有利于未来的可拓展性和可维护性。

例如日志拦截器过滤器事务管理性能统计都是采用AOP面向切面编程

四、Mybatis

Mybatis

4.1 MyBatis框架的优缺点及其适用的场合

优点

与JDBC相比，减少了50%以上的代码量。
MyBatis是易学的持久层框架，小巧并且简单易学。
MyBatis相当灵活，不会对应用程序或者数据库的现有设计强加任何影响，SQL写在XML文件里，从程序代码中彻底分离，降低耦合度，便于统一的管理和优化，并可重用。
提供XML标签，支持编写动态的SQL，满足不同的业务需求。
提供映射标签，支持对象与数据库的ORM字段关系映射。

缺点

SQL语句的编写工作量较大，对开发人员编写SQL的能力有一定的要求。
SQL语句依赖于数据库，导致数据库不具有好的移植性，不可以随便更换数据库。

适用场景

MyBatis专注于SQL自身，是一个足够灵活的DAO层解决方案。对性能的要求很高，或者需求变化较多的项目，例如Web项目，那么MyBatis是不二的选择。

4.2 mapper接口与xml关联原理？

mapper接口是不能重载的。而且必须做到全限名。不能存在多个文件。

mapper接口的工作原理就是通过jdk动态代理的形式。mybatis运行时，会使用动态代理的形式生成对象MapperProxy，代理对象会拦截接口方法，也就是执行invoke函数，invoke函数内操作也就是直接操作sqlsession。

继续利用statement（执行接口所对于的xml内函数的ID）拿到mapperstatement对象，通过执行器Executor去执行具体的SQL执行结果返回

4.3 如何进行分页？分页原理？

使用RowBouds对象进行分页。它是针对ResultSet的内存分页的。mybatis分页原理的核心点是内存分页，而不是物理分页。所以性能是比较快的。

分页插件的实现原理就是一切还是老规矩执行，只是在插件内会有一个拦截的操作。拦截之后会重写SQL ，根据dialect方言，添加对应的物理分页语句和物理分页参数。

4.4 延迟加载（懒加载）

使用CGLIB创建目标对象的代理对象，当调用目标方法时，进入拦截器方法，比如调用a.getB().getName()，拦截器invoke()方法发现a.getB()是null值，那么就会单独发送事先保存好的查询关联B对象的sql，把B查询上来，然后调用a.setB(b)，于是a的对象b属性就有值了，接着完成a.getB().getName()方法的调用。这就是延迟加载的基本原理。

4.5 一级，二级缓存？

一级缓存是存在session中，如果经过flush或者close,Cache将会被清空。close会释放PerpetualCache对象。导致对象不可用。clearCache会清空缓存但是PerpetualCache对象还是可用的。执行任何一个insert，update，delete，都会清空一级缓存

二级缓存是Application级别的缓存，它可以提高对数据库查询的效率，以提高应用的性能。默认情况是不开启二级缓存的,二级缓存是全局缓存。

何时命中缓存？

传入statementId，查询时要求的结果集中的结果范围，传递给java.sql.Statement要设置的参数值，sql字符串（执行过程中，如果满足以上条件，则命中缓存，虽然是多次执行SQL 但是只会执行一遍，第一遍过后的数据会从缓存中取）

执行流程？

一级缓存：client=>Executor=>Database 这三方进行通信，利用Executor与Local Cache进行数据缓存处理

二级缓存：client=>CachingExcutor=>Executor=>Database 这四方通信，利用CachingExcutor与缓存集群Configuration进行通信，缓存子集是Mapper namespace

4.6 不同的映射文件中，ID是否可以重复？原理？

不同的xml映射文件的寻址方式就是 namespace + id。存储的格式是Map<String, MapperStatement>。如果namespace相同的话，只根据id无法判断是哪个键值对，所以多个xml文件不同的命名空间是可以允许相同的id的，反之不允许

五、计算机基础与网络

计算机组成与网络

5.1 TCP和HTTP的区别

TCP是 传输层协议，定义数据传输和连接方式的规范。握手过程中传送的包里不包含数据，三次握手完毕后，客户端与服务器才正式开始传送数据。

HTTP 超文本传送协议(Hypertext Transfer Protocol )是 应用层协议，定义的是传输数据的内容的规范。

HTTP协议中的数据是利用TCP协议传输的，特点是客户端发送的每次请求都需要服务器回送响应，它是TCP协议族中的一种，默认使用 TCP 80端口。

好比网络是路，TCP是跑在路上的车，HTTP是车上的人。每个网站内容不一样，就像车上的每个人有不同的故事一样。

5.2 是否了解操作系统，进程和线程的区别，资源争端怎办？

进程和线程都不是一个东西，无法办法进行比较。

这里可以举一个例子，进程就是QQ，线程就是QQ中的一个发送模块。

一个线程只可以属于一个进程，但是一个进程可以包含多个线程

资源争端也可以说是死锁

死锁，是指多个进程在运行过程中因争夺资源而造成的一种僵局，当进程处于这种僵持状态时，若无外力作用，它们都将无法再向前推进。

例如：系统中只有一台打印机，可供进程 A 使用，假定 A 已占用了打印机，若 B 继续要求打印机打印将被阻塞。

系统中的资源可以分为两类：

可剥夺资源：是指某进程在获得这类资源后，该资源可以再被其他进程或系统剥夺，CPU 和主存均属于可剥夺性资源；
不可剥夺资源，当系统把这类资源分配给某进程后，再不能强行收回，只能在进程用完后自行释放，如磁带机、打印机等。

如何解决死锁？

资源剥夺：挂起某些死锁进程，并抢占它的资源，将这些资源分配给其他死锁进程（但应该防止被挂起的进程长时间得不到资源）；
撤销进程：强制撤销部分、甚至全部死锁进程并剥夺这些进程的资源（撤销的原则可以按进程优先级和撤销进程代价的高低进行）；
进程回退：让一个或多个进程回退到足以避免死锁的地步。进程回退时自愿释放资源而不是被剥夺。要求系统保持进程的历史信息，设置还原点。

5.3 TCP三次握手，丢包如何解决？

TCP三次握手在发送数据包时，如果出现异常，主要分下面三种情况

A发给B的syn中途被丢，没有到达B
B发给A的syn+ack中途被丢，没有到达A
A发给B的ack中途被丢，没有到达B

A发给B的syn中途被丢，没有到达B

A会周期性超时重传，直到收到B的确认

B发给A的syn+ack中途被丢，没有到达A

B会周期性的超时重传，直到收到A的确认

A发给B的ack中途被丢，没有到达B

A发完ack，单方面会认为TCP为Established状态，而B显然认为TCP为Active状态：

假定此时双方都没有数据发送，B会周期性超时重传，直到收到A的确认，收到之后B的TCP连接也会Established状态，双向可以发包
假定此时A有数据发送，B收到A的Data+ACK，自然会切换为Established状态，并接受A的Data
假定B有数据发送，数据发送不了，会周期性超时重传syn+ack，直到收到A的确认才可以发送数据。

客户端syn包超时重传的最大次数是由 tcp_syn_retries 决定的。默认为5次。

5.4 TCP拥塞控制详细说一下？

原因是有可能整个网络环境特别差，容易丢包，那么发送端就应该注意了。

主要用三种方法：

慢启动阈值 + 拥塞避免
快速重传
快速恢复

慢启动阈值 + 拥塞避免

对于拥塞控制来说，TCP 主要维护两个核心状态：

拥塞窗口（cwnd）
慢启动阈值（ssthresh）

在发送端使用拥塞窗口来控制发送窗口的大小。

然后采用一种比较保守的慢启动算法来慢慢适应这个网络，在开始传输的一段时间，发送端和接收端会首先通过三次握手建立连接，确定各自接收窗口大小，然后初始化双方的拥塞窗口，接着每经过一轮 RTT（收发时延），拥塞窗口大小翻倍，直到达到慢启动阈值。

然后开始进行拥塞避免，拥塞避免具体的做法就是之前每一轮 RTT，拥塞窗口翻倍，现在每一轮就加一个。

快速重传

在 TCP 传输过程中，如果发生了丢包，接收端就会发送之前重复 ACK，比如第 5 个包丢了，6、7 达到，然后接收端会为 5，6，7 都发送第四个包的 ACK，这个时候发送端受到了 3 个重复的 ACK，意识到丢包了，就会马上进行重传，而不用等到 RTO （超时重传的时间）

选择性重传：报文首部可选性中加入 SACK 属性，通过 left edge 和 right edge 标志那些包到了，然后重传没到的包

快速恢复

如果发送端收到了 3 个重复的 ACK，发现了丢包，觉得现在的网络状况已经进入拥塞状态了，那么就会进入快速恢复阶段：

会将拥塞阈值降低为拥塞窗口的一半
然后拥塞窗口大小变为拥塞阈值
接着拥塞窗口再进行线性增加，以适应网络状况

六、RocketMQ

RocketMQ

6.1 消息队列用过哪些？消息队列的作用？

常用的消息队列大概有这四种

Kafka：吞吐量大概在10 万级，高吞吐，支持topic。一般配合大数据类的系统来进行实时数据计算、日志采集等场景。延迟在 ms 级以内。它的可用性非常高，分布式，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用。可以做到0丢失
RocketMQ：10 万级，支撑高吞吐，支持topic。延迟在ms 级，可用性非常高，分布式架构。经过参数优化配置，可以做到 0 丢失
ActiveMQ：吞吐量大概在万级，延迟在ms级。可用性高，基于主从架构实现高可用。有较低的概率丢失数据。
RabbitMQ：吞吐量大概在万级，延迟在微秒级，这是 RabbitMQ 的一大特点，延迟最低。可用性高，基于主从架构实现高可用。基本不丢数据

优点与作用

解耦：以前都是接入系统都是改代码调用。现在直接MQ的形式，一个只管发，一个只管接。
异步：程序处理完本职工作之后，把数据写入MQ就可以让他在后台自动执行，就不需要等他执行完再返回了。
削峰：和异步差不多，源源不断的请求堆积在这里，就可以把所有的请求都打到MQ，由MQ按顺序执行即可。

缺点

系统可用性降低：俗话说，写的越多，出错的几率越大。
系统复杂度提高：原本调用就够了，现在要接入MQ，还要做一些参数的配置，性能调优配置。加大的开发难度。
一致性问题：宕机后MQ未处理的消息以及已处理的消息，和数据库，和Redis的数据一致性问题

我的技术选型是RocketMQ，主要原因如下

吞吐量在10万级，延迟低，可以做到0丢失
可用性很高，体量到了一定级别，可以采用分布式架构进行扩展
底层是用Java开发的，故障排查，二次开发也是非常友好的（Java程序员）
阿里开源的，社区较为活跃，出现问题也方便解决

6.2 如何保证消息不丢失？

聊到消息一致性，可靠性传输，我们可以从问题的根源入手。我先列举一些容易出问题的故障点

生产阶段：在这个阶段，从消息在 Producer 创建出来，经过网络传输发送到 Broker 端。
存储阶段：在这个阶段，消息在 Broker 端存储，如果是集群，消息会在这个阶段被复制到其他的副本上。
消费阶段：在这个阶段，Consumer 从 Broker 上拉取消息，经过网络传输发送到 Consumer 上。

生产阶段

在生产阶段，消息队列通过最常用的请求确认机制，来保证消息的可靠传递：当你的代码调用发消息方法时，消息队列的客户端会把消息发送到 Broker，Broker 收到消息后，会给客户端返回一个确认响应，表明消息已经收到了。客户端收到响应后，完成了一次正常消息的发送。

只要 Producer 收到了 Broker 的确认响应，就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送确认响应后，会自动重试，如果重试再失败，就会以返回值或者异常的方式告知用户。

你在编写发送消息代码时，需要注意，正确处理返回值或者捕获异常，就可以保证这个阶段的消息不会丢失

存储阶段

在存储阶段正常情况下，只要 Broker 在正常运行，就不会出现丢失消息的问题，但是如果 Broker 出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。

如果对消息的可靠性要求非常高，可以通过配置 Broker 参数来避免因为宕机丢消息。

对于单个节点的 Broker，需要配置 Broker 参数，在收到消息后，将消息写入磁盘后再给 Producer 返回确认响应，这样即使发生宕机，由于消息已经被写入磁盘，就不会丢失消息，恢复后还可以继续消费。例如，在 RocketMQ 中，需要将刷盘方式 flushDiskType 配置为 SYNC_FLUSH 同步刷盘。

集群我不会，后续再更新。

消费阶段

消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递，客户端从 Broker 拉取消息后，执行用户的消费业务逻辑，成功后，才会给 Broker 发送消费确认响应。如果 Broker 没有收到消费确认响应，下次拉消息的时候还会返回同一条消息，确保消息不会在网络传输过程中丢失，也不会因为客户端在执行消费逻辑中出错导致丢失。

你在编写消费代码时需要注意的是，不要在收到消息后就立即发送消费确认，而是应该在执行完所有消费业务逻辑之后，再发送消费确认。

消息丢失检测

前期代码健壮性不友好的情况，可以在拦截器里编写日志输出，把消费的id号记录下来。

生产者，生产一条就记录一条
消费者，消费一条就记录一条

这样这样两边对照就可以把丢失的id号定位出来。也可以通过分布式链路追踪系统 扯远了，以后再说吧

6.3 如何保证消息顺序消费？

一个topic下有多个队列，为了保证发送有序，RocketMQ提供了MessageQueueSelector队列选择机制，他有三种实现:

我们可使用Hash取模法，让同一个订单发送到同一个队列中，再使用同步发送，只有同个订单的创建消息发送成功，再发送支付消息。这样，我们保证了发送有序。

RocketMQ的topic内的队列机制,可以保证存储满足FIFO（First Input First Output 简单说就是指先进先出）,剩下的只需要消费者顺序消费即可。

RocketMQ仅保证顺序发送，顺序消费由消费者业务保证!!!

这里很好理解，一个订单你发送的时候放到一个队列里面去，你同一个的订单号Hash一下是不是还是一样的结果，那肯定是一个消费者消费，那顺序是不是就保证了？

参考于敖丙的CSDN

6.4 如何处理消息过程中的重复消费？

采用 幂等性

幂等性是一个数学上的概念，它是这样定义的：如果一个函数 f(x) 满足：f(f(x)) = f(x)，则函数 f(x) 满足幂等性。

这里被扩展到计算机领域，被广泛的应用于多次执行产生的影响均与一次执行的影响相同

使用同样的参数，对它进行多次调用和一次调用，对系统产生的影响是一样的。所以，对于幂等的方法，不用担心重复执行会对系统造成任何改变。

数据库的唯一约束实现幂等
为更新的数据设置前置条件
记录并检查操作

数据库的唯一约束实现幂等

我先举一个我自己系统的例子：用户在充值账号余额时，会产生一个账单ID。

我们在实现唯一约束的时候就可以重新创建一个表。伪代码如下

create table aaa(
    id  bigint(15) not null comment '约束id',
    user_id bigint(15) not null comment '用户id',
    bill_id bigint(15) not null comment '账单id',
    money decimal(10,2) not null comment '充值金额',
    PRIMARY KEY (`id`) USING BTREE,
    KEY `adasdasdas` (`user_id`,`bill_id`),  -- 唯一约束  用户di和账单id
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='账单约束表';

这样，我们消费消息的逻辑可以变为：“在转账流水表中增加一条转账记录，然后再根据转账记录，异步操作更新用户余额即可。”在转账流水表增加一条转账记录这个操作中，由于我们在这个表中预先定义了“账户 ID 转账单 ID”的唯一约束，对于同一个转账单同一个账户只能插入一条记录，后续重复的插入操作都会失败，这样就实现了一个幂等的操作。我们只要写一个 SQL，正确地实现它就可以了。

基于这个思路，不光是可以使用关系型数据库，只要是支持类似“INSERT IF NOT EXIST”语义的存储类系统都可以用于实现幂等，比如，你可以用 Redis 的 SETNX 命令来替代数据库中的唯一约束，来实现幂等消费。

参考李玥老师的消息队列高手课思想

为更新的数据设置前置条件

在更新数据时，我们可以设置一个更新前的值，如下图。

这里可以加一个充值前金额，这里因为我的体量，并发不大，暂时没加，后面我会根据老板的要求再加的。

如果有重复订单打过来，那我就可以计算充值前的金额，以及当前的付款金额。来付款来实现幂等性。

也可以通过版本号控制，每次更数据前，比较当前数据的版本号是否和消息中的版本号一致，如果不一致就拒绝更新数据，更新数据的同时将版本号 +1，一样可以实现幂等更新。

在修改数据记录并检查操作

可以采用Token，UUID的方式实现幂等性。这种方式是通用性比较强的。实现的思路特别简单：在执行数据更新操作之前，先检查一下是否执行过这个更新操作。

具体的实现方法是，在发送消息时，给每条消息指定一个全局唯一的 ID，消费时，先根据这个 ID 检查这条消息是否有被消费过，如果没有消费过，才更新数据，然后将消费状态置为已消费。

七、Redis

Redis

7.1 Redis项目中如何用的？用来做什么？

Redis在项目中主要用于缓存。分布式锁，限流，过滤器。也可以适当的充当小型的消息队列。

最核心的还是他的缓存。

众所周知，Redis是单线程+多路复制机制（伪多线程）。使Redis的性能非常的高。

主要用于存储热点数据，用户身份的有效性。

这里可以展开Redis单线程为什么那么快展开介绍一下。可以参考下列文章的第二部分

聊聊Redis面试题

根本不同的场景可以跟面试官都聊一下。比如用于缓存时，

如何解决数据过期时间问题
Redis过期机制（Java中的GC）
Redis过期策略的六大策略。
由六大策略引进LRU算法，LFU算法展开介绍一下
数据修改后，MySQL与Redis如何做好数据同步
Redis底层的数据结构是由什么实现的。
体量增到一定量级时，布隆过滤器的诞生。作用，原理可以介绍一下。

比如用于消息队列时，

消息队列，发布，订阅模式的介绍
Redis用于消息队列时的优缺点

比如用于身份校验时，

数据结构的选用。可以对Map，List，string 做一个数据结构的性能分析

上述的技术问题，在这些【Redis技术文章】中都可以找到答案

7.2 Redis挂了会发生什么？高可用，哨兵模式？

发生什么

Redis挂了，网站能不能访问主要取决于是否在Redis上存了校验Token以及是否在程序里做了try catch异常捕获。

如果存在Token。网站直接会崩溃

如果代码中没有处理好相应的Redis无响应代码也会崩溃。

上述问题可以通过try-catch代码块来解决。但是这样的话就会造成代码臃肿，最好的解决办法就是采用Redis集群的方式，单台Redis挂了，可以通过另外几台继续提供支撑服务。

哨兵模式

其中一台挂了，或者主库挂了，肯定是要涉及到选主的或者去从操作的。由这个问题可以抛出来哨兵模式的技术链，只要扯到Redis挂了，必提哨兵。很多时候不需要等面试官问，我们可以自由先发挥一些。这样面试官也会对你印象深刻一些。

哨兵主要负责的就是三个任务：监控、选主（选择主库）和通知。

监控：哨兵会周期性的给所有的主从库发送ping命令，监测它们是否处于在线运行状态，

如果从库没有响应ping命令，哨兵就会把从库标记为下线状态。
如果主库没有响应ping命令，哨兵就会判断主库下线并且开始自动切换主库。

选主：主库挂了之后，哨兵就需要在很多从库中，按照一定的规则选择出一个从库实例，把它作为新的主库。

通知：主库重新诞生之后，哨兵就会通知从库，告诉它们新主库的信息。让他们执行replicaof 命令。重新建立之后开始数据同步流程。

一系列流程之后，主库出现了，从库也正常了。一切又回到了出故障之前的那个状态！

从库标记下线的话，比较简单一些，这里介绍一下选主的规则

哨兵选主库时，一般都称为 “筛选+打分”

筛选：（网络波动）除了检查从库当前的在线状态还要判断它网络的连接状态。如果从库和主库响应过慢，并且超过了一定的阈值，那么肯定是不能选择该从库充当我们的主库的。因为一旦该从库选择主库，一旦在后续的写入操作，数据同步操作中网络波动大，或者直接断开连接了，我们还需要重新做一下选择，通知，同步等。这样性能是非常低效的！

打分：（择偶标准）主要有三点如下

优先级最高的从库得分高：用户可以通过 slave-priority 配置项，给不同的从库设置不同的优先级，比如不同的从库中的内存配置，CPU配置等
同步进度：一般选择一个从库为主库，如果我们从库的数据同步进度更接近与前主库，那么从库切换成主库之后，数据同步的时间消耗更低。性能会更好一些。
ID 号小的从库得分高。（Redis的默认规定没啥好说的）。在优先级和复制进度都相同的情况下，ID 号最小的从库得分最高，会被选为新主库。

这里有一个面试高频题，就不展开了，有点多。直接上一个参考链接，有兴趣的可以了解一下。

哨兵挂了，主从还能切换吗

7.3 Redis缓存击穿、缓存雪崩、缓存穿透讲讲？

雪崩

雪崩这个东西，见名思意。我们可以理解成一大堆雪冲了过来，一面墙无法抵挡猛烈的攻击。于是冲垮了墙，直接冲向了你们家的卧室！

一般造成缓存雪崩主要是有如下几个原因

发大洪水：系统设计问题，明明有100万流量，程序只设计了10万
洪水从天而降的意外：缓存key刚好大面积失效，过期时间设计的不合理
被瓦匠工糊弄了一下墙倒了：Redis实例宕机

我们在平时解决时，可以避免缓存写入的时间大面积相同，可以在后面加一个随机函数，让过期时间分布的频段多一些。还可以通过服务降级来解决缓存雪崩。

比如在去年新冠疫情的那会。严查所有过往的路人，一旦有咳嗽，发烧一律不予通行。如果没有咳嗽，发烧等情况还持有体检报告的可以回家自行隔离。

在Redis中也是同样道理，如果访问的是核心数据，我们可以放行，如果是访问附加属性我们可以直接返回初始数据，或者网络波动问题。这样就可以过滤一部分附加属性的请求了。

还有一种情况就是，洪水还没来，墙自己倒了。你看这不是赤裸裸的求干吗，你这不就是挑衅洪水的嘛。

一般为了系统业务能正常运行，我们会提前最好做好如下应对措施

实现服务熔断
实现请求限流机制。
高可用集群

服务熔断的话我们可以理解成，为了防止引发连锁反应（积分服务挂了，还能影响订单嘛）我们关掉了用户的积分服务。等修复成功之后再重新开启服务。这样就可以避免其他服务受此牵连。

在业务系统运行时，我们可以监测 Redis 缓存所在机器和数据库所在机器的负载指标，例如每秒请求数、CPU 利用率、内存利用率等。如果我们发现 Redis 缓存实例宕机了，而数据库所在机器的负载压力突然增加（例如每秒请求数激增），此时，就发生缓存雪崩了。大量请求被发送到数据库进行处理。我们可以启动服务熔断机制，暂停业务应用对缓存服务的访问，从而降低对数据库的访问压力

服务限流的话，我们可以理解成早晨上班的警察道路调配。一个路口是流入量是不变的，如果我们想正常运行，就必须把流入速率慢下来。

回到系统的话就是每秒1万个请求，限流之后，每秒1千个请求。再多的请求我们拒之门外排队等候。

高可用集群 的话也算是提前预防了。就好比在双十一或者流量超级春运的时候。流量超级大。我们提前在节日之前把对应的机器设施架设起来。一旦大屏面板监测到大批流量引入我们可以开启备选方案，通过增加机器来解决并发需求。这样也可以达到节省硬件成本的需求。

击穿

缓存击穿主要就是 热点数据失效 。双十一期间如果榜一的商品缓存失效了，恐怕就有悲剧了。一时间所有的请求都打到的数据库上。这是由于热点数据的过期时间设计不符导致的。

我们一般对这类数据会进行提前预热，比如热榜前100的数据，我们会预热30分钟。这样在秒杀的时候，就不会造成在短时间内大量请求打入数据库了。

穿透

缓存穿透顾名思义，就是在玩刺剑时，攒足力气，直冲一处。

回到系统中是这样的意思。黑客在黑入我们系统时，往往会猜想一些缓存中没有的数据，使大量请求打到数据库，造成缓存穿透。当下次再次请求时，缓存中还是没有查询到，因为从数据库查询时，本来就没有所以也无法写入缓存。

我们的解决方案就是：不管数据库是否存在当前数据，我们都缓存的一个key，给这个key的value中写入一个null。

八、项目方案

项目方案

8.1 问秒杀项目：介绍一下你对项目高并发的理解；

高并发意味着大流量，需要运用技术手段抵抗流量的冲击，这些手段好比操作流量，能让流量更平稳地被系统所处理，带给用户更好的体验。

我们常见的高并发场景有：淘宝的双11、春运时的抢票、微博大V的热点新闻等。除了这些典型事情，每秒几十万请求的秒杀系统、每天千万级的订单系统、每天亿级日活的信息流系统等，都可以归为高并发。

很显然，上面谈到的高并发场景，并发量各不相同，那到底多大并发才算高并发呢？

不能只看数字，要看具体的业务场景。不能说10W QPS的秒杀是高并发，而1W QPS的信息流就不是高并发。信息流场景涉及复杂的推荐模型和各种人工策略，它的业务逻辑可能比秒杀场景复杂10倍不止。因此，不在同一个维度，没有任何比较意义。
业务都是从0到1做起来的，并发量和QPS只是参考指标，最重要的是：在业务量逐渐变成原来的10倍、100倍的过程中，你是否用到了高并发的处理方法去演进你的系统，从架构设计、编码实现、甚至产品方案等维度去预防和解决高并发引起的问题？而不是一味的升级硬件、加机器做水平扩展。

8.2 库存超卖如何解决的？(商城类项目)

库存超卖问题一直是电商系统中的一个热点话题。库存超卖无非就是对扣减库存的把控。下面我们从三个方面分析一下。

下单减库存
付款减库存
预扣减库存（失效）

下单减库存

下单减库存，这种方式很好理解。用户下单后就立马扣减库存，但是有一个问题就是，有些用户在 提交订单之后，未必就会付款 。

这样就会存在恶心下单，然后全都不付款。如果在秒杀活动的时候，遇到这种情况，这个活动就等于无效了。显然不是一个好的方案。

付款减库存

付款减付款，用户在下单后，不会扣减库存，只有在付款的时候才会减少库存。这样的问题就是，秒杀那一刻冲进来的流量。会 有部分人已下单但是付款失败 的情况。付款失败是因为在扣减库存的时候校验为0了

付款减库存，还会造成，下单量大于实际库存量。付款那里如果处理不好的话也会造成 库存超卖 的情况

预扣减库存（失效）

预扣减库存也是现在电商平台，外卖平台，打车平台常用的一种模式。下单后如果5-10分钟后仍不付款，系统就会自动取消。释放对应的库存。

问题并没有从根本上解决

虽然预扣减库存方式极大的解决了库存问题，但是仍然有漏洞。

比如10分钟自动取消，在双十一秒杀期间，如果有人恶意下单后在第9分钟的时候把订单取消了，取消后就立马重新下单。还会占用相应的库存。

解决方案

针对恶意用户下单的情况，我这里简单罗列了如下几种解决方案：

我们可以为经常提交订单之后不付款的用户添加对应的标签，当这些用户下单时，进行特殊处理，例如不扣减库存等（具体可以根据需求确定）。
在秒杀期间，为商品设置同一个人的最大购买件数，比如最多购买2件。
对不付款重复下单的操作进行限制，例如，对同一商品下单时，首先校验当前用户是否存在未付款的订单，并且订单中的商品与再次下单的商品是同一款商品，则提示先让用户付款后再提交订单等。

针对库存超卖的情况，我这里简单罗列了如下几种解决方案：

通过补货解决。
用户下单时提示库存不足。

秒杀系统如何扣减库存？

在真正的高并发、大流量场景下，大部分秒杀系统会采用 下单减库存 的方式。

在下单扣减库存的业务场景中，需要保证大流量、高并发下商品的库存不能为负。

这里，我们可以通过如下方案解决商品库存不能为负的问题、

在扣减库存后，通过在应用程序的事务中判断商品库存是否为负数，如果变成了负数，则回滚事务不再扣减库存。

在数据库中设置库存字段为无符号整数，从数据库层面保证无法出现负数的情况。