Java高并发 -- 并发扩展

作者: sunhaiyu | 来源:发表于2019-04-20 16:10 被阅读69次

Java高并发 -- 并发扩展

主要是学习慕课网实战视频《Java并发编程入门与高并发面试》的笔记

死锁

死锁是指两个或两个以上的事务在执行过程中，因争夺锁资源而造成的一种互相等待的现象，若无外力作用两个事务都无法推进，这样就产生了死锁。死锁的四个必要条件：

互斥条件：即任何时刻，一个资源只能被一个进程使用。其他进程必须等待。
请求和保持条件：即当资源请求者在请求其他的资源的同时保持对原有资源的占有且不释放。
不剥夺条件：资源请求者不能强制从资源占有者手中夺取资源，资源只能由资源占有者主动释放。
环路等待条件：比如A占有B在等待的资源（B等待A释放），B占有A在等待的资源（A等待B释放）。多个进程循环等待着相邻进程占用着的资源。

是指两个或两个以上的线程在执行过程中，互相占用着对方想要的资源但都不释放，造成了互相等待，结果线程都无法向前推进。

死锁的检测：可以采用等待图（wait-for gragh）。采用深度优先搜索的算法实现，如果图中有环路就说明存在死锁。

解决死锁：

破环锁的四个必要条件之一，可以预防死锁。
加锁顺序保持一致。不同的加锁顺序很可能导致死锁，比如哲学家问题：A先申请筷子1在申请筷子2，而B先申请筷子2在申请筷子1，最后谁也得不到一双筷子（同时拥有筷子1和筷子2）
撤消或挂起进程，剥夺资源。终止参与死锁的进程，收回它们占有的资源，从而解除死锁。

举个简单的例子

public class DeadLock implements Runnable {

    public static Object fork1 = new Object();
    public static Object fork2 = new Object();

    private String name;
    private Object tool;

    public DeadLock(Object o) {
        this.tool = o;
        if (tool == fork1) {
            this.name = "哲学家A";
        }
        if (tool == fork2) {
            this.name = "哲学家B";
        }
    }

    @Override
    public void run() {
        if (tool == fork1) {
            synchronized (fork1) {
                try {
                    System.out.println(name+"拿到了一个叉子");
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (fork2) {
                    System.out.println(name+"拿到两个叉子了");
                }
            }
        }

        if (tool == fork2) {
            synchronized (fork2) {
                try {
                    System.out.println(name+"拿到了一个叉子");
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (fork1) {
                    System.out.println(name+"拿到两个叉子了");
                }
            }
        }
    }

    public static void main(String[] args) {
        DeadLock a = new DeadLock(fork1);
        DeadLock b = new DeadLock(fork2);

        Thread t1 = new Thread(a);
        Thread t2 = new Thread(b);
        t1.start();
        t2.start();

    }
}

运行上面这段程序，会输出

哲学家B拿到了一个叉子
哲学家A拿到了一个叉子

然后程序就进入了死循环，因为哲学家A在等B手里的叉子，哲学家B也在等A手上的叉子，但是他俩谁都不肯释放。

HashMap和ConcurrentHshMap

HashMap

HashMap的底层使用数组+链表/红黑树实现。

transient Node<K,V>[] table;这表示HashMap是Node数组构成，其中Node类的实现如下，可以看出这其实就是个链表，链表的每个结点是一个<K,V>映射。

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    V value;
    Node<K,V> next;
}

HashMap的每个下标都存放了一条链表。

常量/变量定义

/* 常量定义 */

// 初始容量为16
static final int DEFAULT_INITIAL_CAPACITY = 1 << 4;
// 最大容量
static final int MAXIMUM_CAPACITY = 1 << 30;
// 负载因子，当键值对个数达到DEFAULT_INITIAL_CAPACITY * DEFAULT_LOAD_FACTOR会触发resize扩容 
static final float DEFAULT_LOAD_FACTOR = 0.75f;
// 当链表长度大于8，且数组长度大于MIN_TREEIFY_CAPACITY，就会转为红黑树
static final int TREEIFY_THRESHOLD = 8;
// 当resize时候发现链表长度小于6时，从红黑树退化为链表
static final int UNTREEIFY_THRESHOLD = 6;
// 在要将链表转为红黑树之前，再进行一次判断，若数组容量小于该值，则用resize扩容，放弃转为红黑树
// 主要是为了在建立Map的初期，放置过多键值对进入同一个数组下标中，而导致不必要的链表->红黑树的转化，此时扩容即可，可有效减少冲突
static final int MIN_TREEIFY_CAPACITY = 64;

/* 变量定义 */

// 键值对的个数
transient int size;
// 键值对的个数大于该值时候，会触发扩容
int threshold;
// 非线程安全的集合类中几乎都有这个变量的影子，每次结构被修改都会更新该值，表示被修改的次数
transient int modCount;

关于modCount的作用见这篇blog

在一个迭代器初始的时候会赋予它调用这个迭代器的对象的modCount，如果在迭代器遍历的过程中，一旦发现这个对象的modCount和迭代器中存储的modCount不一样那就抛异常。
Fail-Fast机制：java.util.HashMap不是线程安全的，因此如果在使用迭代器的过程中有其他线程修改了map，那么将抛出ConcurrentModificationException，这就是所谓fail-fast策略。这一策略在源码中的实现是通过modCount域，modCount顾名思义就是修改次数，对HashMap内容的修改都将增加这个值，那么在迭代器初始化过程中会将这个值赋给迭代器的expectedModCount。在迭代过程中，判断modCount跟expectedModCount是否相等，如果不相等就表示已经有其他线程修改了Map。

注意初始容量和扩容后的容量都必须是2的次幂，为什么呢?

hash方法

先看散列方法

static final int hash(Object key) {
    int h;
    return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}

HashMap的散列方法如上，其实就是将hash值的高16位和低16位异或，我们将马上看到hash在与n - 1相与的时候，高位的信息也被考虑了，能使碰撞的概率减小，散列得更均匀。

在JDK 8中，HashMap的putVal方法中有这么一句

if ((p = tab[i = (n - 1) & hash]) == null)
    tab[i] = newNode(hash, key, value, null);

关键就是这句(n - 1) & hash，这行代码是把待插入的结点散列到数组中某个下标中，其中hash就是通过上面的方法的得到的，为待插入Node的key的hash值，n是table的容量即table.length，2的次幂用二进制表示的话，只有最高位为1，其余为都是0。减去1，刚好就反了过来。比如16的二进制表示为10000，减去1后的二进制表示为01111，除了最高位其余各位都是1，保证了在相与时，可以使得散列值分布得更均匀（因为如果某位为0比如1011，那么结点永远不会被散列到1111这个位置），且当n为2的次幂时候有(n - 1) & hash == hash % n, 举个例子，比如hash等于6时候，01111和00110相与就是00110，hash等于16时，相与就等于0了，多举几个例子便可以验证这一结论。最后来回答为什么HashMap的容量要始终保持2的次幂

使散列值分布均匀
位运算的效率比取余的效率高

注意table.length是数组的容量，而transient int size表示存入Map中的键值对数。

int threshold表示临界值，当键值对的个数大于临界值，就会扩容。threshold的更新是由下面的方法完成的。

static final int tableSizeFor(int cap) {
    int n = cap - 1;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}

该方法返回大于等于cap的最小的二次幂数值。比如cap为16，就返回16，cap为17就返回32。

put方法

put方法主要由putVal方法实现：

如果没有产生hash冲突，直接在数组tab[i = (n - 1) & hash]处新建一个结点；
否则，发生了hash冲突，此时key如果和头结点的key相同，找到要更新的结点，直接跳到最后去更新值
否则，如果数组下标中的类型是TreeNode，就插入到红黑树中
如果只是普通的链表，就在链表中查找，找到key相同的结点就跳出，到最后去更新值；到链表尾也没有找到就在尾部插入一个新结点。接着判断此时链表长度若大于8的话，还需要将链表转为红黑树（注意在要将链表转为红黑树之前，再进行一次判断，若数组容量小于64，则用resize扩容，放弃转为红黑树）

get方法

get方法由getNode方法实现：

如果在数组下标的链表头就找到key相同的，那么返回链表头的值
否则如果数组下标处的类型是TreeNode，就在红黑树中查找
否则就是在普通链表中查找了
都找不到就返回null

remove方法的流程大致和get方法类似。

HashMap的扩容，resize()过程？

newCap = oldCap << 1

resize方法中有这么一句，说明是扩容后数组大小是原数组的两倍。

Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
        table = newTab;
        if (oldTab != null) {
            for (int j = 0; j < oldCap; ++j) {
                Node<K,V> e;
                if ((e = oldTab[j]) != null) {
                    oldTab[j] = null;
                    // 如果数组中只有一个元素，即只有一个头结点，重新哈希到新数组的某个下标
                    if (e.next == null)
                        newTab[e.hash & (newCap - 1)] = e;
                    else if (e instanceof TreeNode)
                        ((TreeNode<K,V>)e).split(this, newTab, j, oldCap);
                    else { // preserve order
                        // 数组下标处的链表长度大于1，非红黑树的情况
                        Node<K,V> loHead = null, loTail = null;
                        Node<K,V> hiHead = null, hiTail = null;
                        Node<K,V> next;
                        do {
                            next = e.next;
                            // oldCap是2的次幂，最高位是1，其余为是0，哈希值和其相与，根据哈希值的最高位是1还是0，链表被拆分成两条，哈希值最高位是0分到loHead。
                            if ((e.hash & oldCap) == 0) {
                                if (loTail == null)
                                    loHead = e;
                                else
                                    loTail.next = e;
                                loTail = e;
                            }
                            // 哈希值最高位是1分到hiHead
                            else {
                                if (hiTail == null)
                                    hiHead = e;
                                else
                                    hiTail.next = e;
                                hiTail = e;
                            }
                        } while ((e = next) != null);
                        if (loTail != null) {
                            loTail.next = null;
                            // loHead挂到新数组[原下标]处；
                            newTab[j] = loHead;
                        }
                        if (hiTail != null) {
                            hiTail.next = null;
                            // hiHead挂到新数组中[原下标+oldCap]处
                            newTab[j + oldCap] = hiHead;
                        }
                    }
                }
            }
        }
        return newTab;

举个例子，比如oldCap是16，二进制表示是10000，hash值的后五位和oldCap相与，因为oldCap的最高位（从右往左数的第5位）是1其余位是0，因此hash值的该位是0的所有元素被分到一条链表，挂到新数组中原下标处，hash值该位为1的被分到另外一条链表，挂到新数组中原下标+oldCap处。举个例子：桶0中的元素其hash值后五位是0XXXX的就被分到桶0种，其hash值后五位是1XXXX就被分到桶4中。

ConcurrentHashMap

JDK 7中使用的是分段锁，内部分成了16个Segment即分段，每个分段可以看作是一个小型的HashMap，每次put只会锁定一个分段，降低了锁的粒度：

首先根据key计算出一个hash值，找到对应的Segment
调用Segment的lock方法（Segment继承了重入锁），锁住该段内的数据，所以并没有锁住ConcurrentHashMap的全部数据
根据key计算出hash值，找到Segment中数组中对应下标的链表，并将该数据放置到该链表中
判断当前Segment包含元素的数量大于阈值，则Segment进行扩容（Segment的个数是不能扩容的，但是单个Segment里面的数组是可以扩容的）

多线程put的时候，只要被加入的键值不属于同一个分段，就可以做到真正的并行put。对不同的Segment则无需考虑线程同步，对于同一个Segment的操作才需考虑。

JDK 8中使用了CAS+synchronized保证线程安全，也采取了数组+链表/红黑树的结构。

put时使用synchronized锁住了桶中链表的头结点。

数组的扩容，被问到了我在看吧.....我只知道多个线程可以协助数据的迁移。

有这么一个问题，ConcurrentHashMap，有三个线程，A先put触发了扩容，扩容时间很长，此时B也put会怎么样？此时C调用get方法会怎么样？C读取到的元素是旧桶中的元素还是新桶中的

A先触发扩容，ConcurrentHashMap迁移是在锁定旧桶的前提下进行迁移的，并没有去锁定新桶。

在某个桶的迁移过程中，别的线程想要对该桶进行put操作怎么办？一旦某个桶在迁移过程中了，必然要获取该桶的锁，所以其他线程的put操作要被阻塞。因此B被阻塞。
某个桶已经迁移完成（其他桶还未完成），别的线程想要对该桶进行put操作怎么办？该线程会首先检查是否还有未分配的迁移任务，如果有则先去执行迁移任务，如果没有即全部任务已经分发出去了，那么此时该线程可以直接对新的桶进行插入操作（映射到的新桶必然已经完成了迁移，所以可以放心执行操作）

ConcurrentHashMap的get操作没有加锁，所以可以读取到值，不过是旧桶中的值。

if (finishing) {
    nextTable = null;
    table = nextTab;
    sizeCtl = (n << 1) - (n >>> 1);
    return;
}

从table = nextTable可以看出，当所有数据迁移完成时，才将用nextTab新数组去覆盖旧数组table。所以在A扩容过程中，C读取到的是旧数组中的元素。

扩容

垂直扩容（垂直扩展）：提升单机处理能力（提高硬盘、内存、CPU）
水平扩容（水平扩展）：增加系统成员数（增加服务器个数）

Java高并发 -- 并发扩展