手记

ConcurrentHashMap 原理和源码分析(一)

青春有我
2019-02-22 21:57:48浏览 4365

说实话，之前几个数据结构HashMap、HashTable 复杂度跟ConcurrentHashMap相比简直是小儿科。

少说废话，直接引入正题。

其实有个疑问，既然有了HashMap和HashTable，为什么会需要ConcurrentHashMap？

因为呢，HashMap是线程不安全的，而HashTable虽然是线程安全的，方法都是用Synchronized修饰的，但争夺的都是同一个对象锁，在高并发的情况下，会产生效率低，等待时间长的问题。这个时候，ConcurrentHashMap就荣耀登场了，至于为什么ConcurrentHashMap能解决高并发的情况？下面会详细解释。

ConcurrentHashMap的特性和原理

JDK1.8 跟之前的版本，ConcurrentHashMap的实现变化了很大。以下都是基于JDK1.8的源码和资料。

  1. concurrentHashMap 不支持null的key和value

  2. concurrentHashMap 很多地方使用了cas操作和分段加锁，加锁的最小单位是Hash桶，这使得ConcurrentHashMap效率大大提升。

  3. 数据结构：

数据结构描述

ConcurrentHashMap的数据结构跟HashMap一样：数组 + 链表 + 红黑树。当hash桶的节点数量超过8个，链表就会转化为红黑树，反之，节点数量减少到6就会转化链表

源码分析

1. 重要的常量
// 初始化容量private static final int DEFAULT_CAPACITY = 16;// 加载因子，跟hashmap一样private static final float LOAD_FACTOR = 0.75f;// 如果发现链表长度小于8，会从链表转化成树，跟hashmap一样static final int TREEIFY_THRESHOLD = 8;// 在哈希表扩容时，如果发现链表长度小于 6，则会由树变成链表，跟hashmap一样static final int UNTREEIFY_THRESHOLD = 6;// 最低树化的容量，如果 容量 < MIN_TREEIFY_CAPACITY 会发生一次resize()static final int MIN_TREEIFY_CAPACITY = 64;// 迁移桶的最低数量// 表示扩容中，一个线程的一次任务负责迁移最少16个hash桶// 后面结合代码再理解下private static final int MIN_TRANSFER_STRIDE = 16;// 用于生成每次扩容都唯一的生成戳的数// 后面结合代码再理解下private static final int RESIZE_STAMP_BITS = 16;// 最大的扩容线程的数量private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;// 移位量private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;// 以下是标记几个特殊的节点的hash值，都是负数// ForwardingNode节点，表示该节点正在处于扩容工作，内部有个指针指向nextTablestatic final int MOVED     = -1;// 红黑树的首节点，内部不存key、value，只是用来表示红黑树static final int TREEBIN   = -2; 

// ReservationNode保留节点，// 当hash桶为空时，充当首结点占位符，用来加锁，在compute/computeIfAbsent使用static final int RESERVED  = -3; 
 
// 用于普通节点hash计算// 结合上面三个变量,特殊节点的hash值都是负数，普通节点为正数static final int HASH_BITS = 0x7fffffff;// CPU数量static final int NCPU = Runtime.getRuntime().availableProcessors();
2. 后面常用到的方法：
// 计算hash值// 让高16位 亦或 低16位，再把高的16位置为0
 static final int spread(int h) {        // & HASH_BITS用于把hash值转化为正数
        return (h ^ (h >>> 16)) & HASH_BITS;
 }// 跟hashmap扩容一样，计算出比c大，最小的2次幂的数，如14->16,29->32private static final int tableSizeFor(int c) {        int n = c - 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;
}
// 根据table的长度n生成一个戳，表示要扩容n长度的table，会构造出sizeCtl// RESIZE_STAMP_BITS = 16static final int resizeStamp(int n) {    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

下面是CAS方法，CAS（compareAndSwap） 比较并交换，是无锁操作的重要手段，是一个原子操作

CAS(V, E, N)：如果变量V跟旧的预期值E相同，则修改成新值N，否则什么都不做

// 获取table[i] 得到hash桶首结点static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {   return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}// 更新table[i]，Node链表 或者TreeBinstatic final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,  Node<K,V> c, Node<K,V> v) {   return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}// 修改table[i]static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
   U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
3. 重要的变量
// 存储node节点的数组transient volatile Node<K,V>[] table;// 扩容后的新的table数组，只有在扩容时才有用// nextTable != null，说明在进行扩容private transient volatile Node<K,V>[] nextTable;// 在初始化或resize时控制参数，重要，后面细说private transient volatile int sizeCtl;// 扩容下个表的索引，重要，后面细说
 private transient volatile int transferIndex;// Node节点的数量，根据cas更新的，有可能不准确，需要结合counterCells一起计算private transient volatile long baseCount;// 操作counterCells的自旋锁private transient volatile int cellsBusy;// counterCell表，大小是2的幂数，// 并行计算每个bucket的元素数量，结合baseCount算法出size，下面细说private transient volatile CounterCell[] counterCells;

以上是理解ConcurrentHashMap非常重要的几个变量，其中有几个没有细说

sizeCtl

这个控制参数贯穿了初始化或扩容，而且不同状态下表达不同的含义。

  1. sizeCtl == 0时候，默认情况

  2. sizeCtl == -1 时候，说明table正在初始化

  3. sizeCtl > 0 时候，说明接下来初始化要的初始化容量或者是扩容成功后threadshold的值

  4. sizeCtl < 0 时候，说明正在扩容，而此刻的sizeCtl是怎么来的呢？

// 假如有一个线程，准备加入扩容，下面开始计算sizeCtl int rs = resizeStamp(table长度);if (sizeCtl 未初始化 或准备初始化) { // sizeCtl >= 0
    sizeCtl =  (rs << RESIZE_STAMP_SHIFT) + 2 // RESIZE_STAMP_SHIFT = 16} else { // sizeCtl < 0 ，正在扩容
    sizeCtl = rs + 1; //增加一个扩容的线程}

rs << 16 左移16位，这样低16位都是0；
rs << 16 + 2 应该理解成 rs << 16 + 1 + 1，第一个1表示初始状态，第二个1表示目前有一个线程参与扩容。
sizeCtl分成了高16位，做验证使用，防止扩容重叠；低16位表示 n - 1个线程在参与扩容线程数

transferIndex

这个变量跟扩容迁移有关，原来table扩容新的nextTable，需要多个线程参与节点的迁移。

transferIndex 从 table.length开始，表示需要迁移的桶的数量或者可以说是索引。

每个线程每次进来，如果发现正在扩容并且 transferIndex > 0 的时候，会停止手头的工作，加入帮助扩容，从中分配得到一个 [trasferIndex - stride, transferIndex）区间对应的hash桶的迁移工作，transferIndex 慢慢的减少直至为0。

也有可能是同一个线程负责了多次任务，迁移了多个stride数量hash桶。

下面只是一个示意图，并不表示一次任务只能移动四个hash桶，并且stride的数量是根据CPU数量和tab的数量决定的，最小是MIN_TRANSFER_STRIDE（16）。

示意图

baseCount, cellBusy, counterCells 计数统计

这三个变量跟统计节点数量有关，

  • baseCount用于记录节点的个数

  • cellsBusy是一个只有0和1两个状态的volatile整数，它被当做一个自旋锁，0代表无锁，1代表加锁，只要对counterCells操作，都需要先CAS更新cellsBusy加锁

  • counterCells 是一个辅助baseCount计数的数组，每个counterCell存着部分的节点数量，这样做的目的就是尽可能地减少冲突，看完下面的流程就可以明白了。

  • table节点的数量 = baseCount + counterCells每个cell记录下来的节点数量

总体的原则就是：先尝试更新baseCount，失败再利用CounterCell

  1. 通过CAS尝试更新baseCount ，如果更新成功则完成，如果CAS更新失败会进入下一步；

  2. 线程通过随机数ThreadLocalRandom.getProbe() & (n-1) 计算出在counterCells数组的位置，如果不为null，则CAS尝试在couterCell上直接增加数量，如果失败会进入下一步；

  3. counterCells数组会进行扩容为原来的两倍，继续随机，继续添加；

  4. 最后，table节点的数量 = baseCount + counterCells每个cell记录下来的节点数量

重要的内部类

// ConcurrentHashMap的节点static class Node<K,V> implements Map.Entry<K,V> {        final int hash;        final K key;        volatile V val;        volatile Node<K,V> next;    // 比较节点是否相同：value跟key都相同
    public final boolean equals(Object o) {
            Object k, v, u; Map.Entry<?,?> e;            return ((o instanceof Map.Entry) &&
                    (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
                    (v = e.getValue()) != null &&
                    (k == key || k.equals(key)) &&
                    (v == (u = val) || v.equals(u)));
    }
}
//红黑树的根节点
 static final class TreeBin<K,V> extends Node<K,V> {
        TreeNode<K,V> root;        volatile TreeNode<K,V> first;        volatile Thread waiter;        volatile int lockState；// 当前的锁状态
        static final int WRITER = 1; // 正在写
        static final int WAITER = 2; // 等待写
        static final int READER = 4; // 正在读
        ......
 }
// 红黑树的节点static final class TreeNode<K,V> extends Node<K,V> {
        TreeNode<K,V> parent;  // red-black tree links
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;
}
ForwardingNode

下面还有一个比较特别的节点

ForwardingNode 是临时节点，这个节点会出现在扩容的时候，不存储实际的数据数据。

如果Hash桶被迁移到新的table中，会在旧的table插入一个ForwardingNode临时节点，内部会指向新的table。

当读操作碰到ForwardingNode，会通过ForwardingNode内部的nextTable找到新的table，继续读。

当写操作碰到ForwadingNode，加入帮助扩容。

static final class ForwardingNode<K,V> extends Node<K,V> {    final Node<K,V>[] nextTable;
    ForwardingNode(Node<K,V>[] tab) {        super(MOVED, null, null, null); // hash设置为move，为-1
        this.nextTable = tab;
    }    //重写了Node中的find方法
    Node<K,V> find(int h, Object k) {        // 避免多次碰到ForwardingNode导致递归过深
        outer: for (Node<K,V>[] tab = nextTable;;) {
            Node<K,V> e; int n;            if (k == null || tab == null || (n = tab.length) == 0 ||
                (e = tabAt(tab, (n - 1) & h)) == null)                return null;            for (;;) {                int eh; K ek;                if ((eh = e.hash) == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))                    return e;                if (eh < 0) {                    if (e instanceof ForwardingNode) {// 还碰到ForwardingNode，往下递归接着找
                        tab = ((ForwardingNode<K,V>)e).nextTable;                        continue outer;
                    }                    else
                        return e.find(h, k);
                }                if ((e = e.next) == null)                    return null;
            }
        }
    }
}



作者：三木仔
链接：https://www.jianshu.com/p/c813d2a9a659


相关标签
Android
0人推荐
随时随地看视频
慕课网APP