一, 常量

// 默认初始容量 16
private static final int DEFAULT_CAPACITY = 16;
// 默认并发等级 16
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
// 默认加载因子 0.75
private static final float LOAD_FACTOR = 0.75f;
// 由链表结构转为树结构的阀值 8
static final int TREEIFY_THRESHOLD = 8;
// 由树结构转为链表的阀值 6
static final int UNTREEIFY_THRESHOLD = 6;
// 最小变成树结构的容量 如果容量小于这个值, 就算链表达到8, 也不进行转换
static final int MIN_TREEIFY_CAPACITY = 64;
private static final int MIN_TRANSFER_STRIDE = 16;

二, 变量

// 控制初始化和扩容,
// -1表示初始化, 其他负数为激活的线程数 +1取反
private transient volatile int sizeCtl;


transient volatile Node<K,V>[] table;
private transient volatile Node<K,V>[] nextTable;

sizeCtrl的含义:

1, 为0, 表示数组未初始化, 且数组初始化容量为16
2, 为正数, 如果数组未初始化, 则记录的是数组初始化容量, 如果已经初始化, 记录数组的扩容阀值(数组的初始容量*0.75)
3, 为-1, 表示数组正在初始化
4, 小于0, 又不是-1, 表示正在扩容, -(1+n), 表示有n个线程正在共同完成数组的扩容操作

三, 节点

static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        volatile V val;
        volatile Node<K,V> next;
}
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;
        ... ...
        /**
         * Returns the TreeNode (or null if not found) for the given key
         * starting at given root.
         */
        final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
            .......
    }

四, 初始化

// 空参构造
public ConcurrentHashMap() {   }
// 带一个参数(容量)
public ConcurrentHashMap(int initialCapacity) {
        if (initialCapacity < 0) { throw new IllegalArgumentException();}
        int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                   MAXIMUM_CAPACITY :
                   tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
        // sizeCtl 为 (容量 + 容量/2 + 1) 取2的n次方, 如容量=16, 结果为32
        this.sizeCtl = cap;
    }

 public ConcurrentHashMap(int initialCapacity, float loadFactor) {
        // jdk1.7时, 并发等级此处默认16
        this(initialCapacity, loadFactor, 1);
    }

 public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
        ... ...
        if (initialCapacity < concurrencyLevel)   // Use at least as many bins
            // 初始化容量不小于并发等级
            initialCapacity = concurrencyLevel;   // as estimated threads
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
        // sizeCtl  容量/加载因子 + 1 取2的n次方
        this.sizeCtl = cap;
    }

小结:
1, 数组的初始化长度, 如果没指定, 会在初始化时使用默认值 16, 如果传了容量, 则为(容量 + 容量/2 + 1) 取2的n次方,如果传了容量和加载因子, 则为(容量/加载因子 + 1) 取2的n次方
2, 加载因子值在计算初始长度时使用, 计算扩容时用不到. 和HashMap有些不同.

五, 存放

public V put(K key, V value) {
        return putVal(key, value, false);
    }

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
        // key和value都不能为null
        if (key == null || value == null) throw new NullPointerException();
        // hash值会一直为正数 
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            if (tab == null || (n = tab.length) == 0){
                // 1, 表格初始化
                tab = initTable();
            } else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                // 2,所在的节点为null, 则使用cas操作,新建一个节点, 
                // 如果新建成功返回true结束循环, 失败则自旋(执行下一次循环)
                if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            } else if ((fh = f.hash) == MOVED) {
                // 3. 扩容情况下, hash值会等于MOVED, 此时去帮助扩容
                tab = helpTransfer(tab, f);
            } else {
                // 4, 
                V oldVal = null;
                // 4.1 锁住当前节点
                synchronized (f) {
                    // 此时判断是否节点被改变了, 如果改变了则进入下一个循环
                    if (tabAt(tab, i) == f) {               
                        if (fh >= 0) {
                             // 4.1.1 普通节点 -- 链表结构
                            // 依次查看key是否有相同的, 没有则新建一个
                            // 有则判断是否替换, oldval记录旧的value
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                if (e.hash == hash && ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }  else if (f instanceof TreeBin) {
                             // 4.1.2 树节点
                            Node<K,V> p;
                            binCount = 2;
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        // 链表长度达到阀值, 会判断进行转为树结构(总长度大于64)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 大小+1, 并判断是否扩容 
        addCount(1L, binCount);
        return null;
    }

ConcurrentHashMap大多时候是通过cas+自旋(就是循环)操作保证线程的安全性. 第2步中, 只有节点为null, 才会新建节点, 从而跳出循环.

六, 初始化

private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0) {
                // sizeCtrl为负数,说明在扩容或者在初始化
                // 当前线程状态由执行修改为就绪状态,放弃CPU执行权
                Thread.yield(); // lost initialization race; just spin
            } else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                // 执行初始化时,会将sizeCtl设置为-1
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        // n就是初始化数组长度, 没有指定则为默认值16
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        // 初始化后, sizeCtl表示下次扩容的阀值, 就是n*3/4
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

小结:
1, 数组初始化时没有用到锁, 而是用sizeCtl+cas操作来保证线程安全. 当sizeCtl为负数时,表示有线程在初始化, 此时不进行操作,放弃CPU执行权.
2, 数组的初始化长度,如果没有指定, 则默认为16. 注意 长度不等于容量.
3, 初始化后sizeCtl表示扩容阀值, 大小为 数组长度*3/4 ;

七,元素数量的维护

// 内部类
 @sun.misc.Contended static final class CounterCell {
        volatile long value;
        CounterCell(long x) { value = x; }
    }
// 全局变量, 维护一个数量单元的数组,记录table数组中, 每个单元内相连节点个数
private transient volatile CounterCell[] counterCells;

private transient volatile long baseCount;

获取数量:

public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }
// 计算总的元素数量
final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                  // 总数量是由基础数量+每个数量单元内的value值
                    sum += a.value;
            }
        }
        return sum;
    }

添加数量

// x为需要添加的数量, 
// check含义: 小于0, 不检查是否扩容, 小于或等于1, 只在无线程竞争时检查是否扩容
private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
      // countterCells不为null时,进入方法
      // 为null时, 判断baseCount是否改变, 没变直接加上x, 否则进入方法
        if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            // 进入方法条件: counterCells为null或者长度为0或者当前线程对应的单元格为null, 如果这些满足, 表示没有线程竞争, uncontentded=true
            // 如果以上方法不满足, 这比较线程所在单元格的值是否被其他线程改变, 改变了说明有竞争,uncontended=false, 进入方法,没有改变则直接把值改为 +x, 不进入方法 
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                // 添加数量
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            s = sumCount();
        }
        // 判断是否需要扩容
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            // 数量达到了扩容阀值, 执行扩容
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                int rs = resizeStamp(n);
                if (sc < 0) {
                    // sizeCtl小于0 , 表示正在扩容
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        // nt不为null, 协助扩容
                        transfer(tab, nt);
                }
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    // 进行扩容
                    transfer(tab, null);
                s = sumCount();
            }
        }
    }

private final void fullAddCount(long x, boolean wasUncontended) {
        int h;
        // 计算当前线程的随机数,用来获取坐标值
        if ((h = ThreadLocalRandom.getProbe()) == 0) {
            ThreadLocalRandom.localInit();      // force initialization
            h = ThreadLocalRandom.getProbe();
            wasUncontended = true;
        }
        // 为true表示集合满了
        boolean collide = false;                // True if last slot nonempty
        for (;;) {
            CounterCell[] as; CounterCell a; int n; long v;
            if ((as = counterCells) != null && (n = as.length) > 0) {
                // 1, 数组已经初始化
                if ((a = as[(n - 1) & h]) == null) {
                    // 1.1 线程所指向的单元为空
                    if (cellsBusy == 0) {            // Try to attach new Cell
                        // 获取修改权(cellsBusy), 新建元素放入到数组中
                        CounterCell r = new CounterCell(x); // Optimistic create
                        if (cellsBusy == 0 &&
                            U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {                          
                            boolean created = false;
                            try {               // Recheck under lock
                                CounterCell[] rs; int m, j;
                                if ((rs = counterCells) != null &&
                                    (m = rs.length) > 0 &&
                                    rs[j = (m - 1) & h] == null) {
                                    rs[j] = r;
                                    created = true;
                                }
                            } finally {
                                cellsBusy = 0;
                            }
                            // 添加成功,跳出循环,失败(2个线程同时修改), 继续下一次
                            if (created)
                                break;
                            continue;           // Slot is now non-empty
                        }
                    }
                    collide = false;
                }
                else if (!wasUncontended)       // CAS already known to fail
                    // 1.2 线程所致单元格不为空, 有线程竞争,改为无竞争?
                    wasUncontended = true;      // Continue after rehash
                else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))  
                    // 1.3 重试单元格是否完成修改, 成功跳出循环
                    break;
                else if (counterCells != as || n >= NCPU)       
                    // NCPU为服务器的核数 , 小于该值可以进行扩容      
                    collide = false;            // At max size or stale
                else if (!collide)
                    collide = true;
                else if (cellsBusy == 0 &&
                         U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                    try {
                        // 此时数组扩容为2倍,并赋值新数组
                        if (counterCells == as) {// Expand table unless stale
                            CounterCell[] rs = new CounterCell[n << 1];
                            for (int i = 0; i < n; ++i)
                                rs[i] = as[i];
                            counterCells = rs;
                        }
                    } finally {
                        cellsBusy = 0;
                    }
                    collide = false;
                    continue;                   // Retry with expanded table
                }
                h = ThreadLocalRandom.advanceProbe(h);
            }
            else if (cellsBusy == 0 && counterCells == as &&
                     U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                // 2. 第一次进入此处, 初始化数组, 放入x
                boolean init = false;
                try {                           // Initialize table
                    if (counterCells == as) {
                        CounterCell[] rs = new CounterCell[2];
                        rs[h & 1] = new CounterCell(x);
                        counterCells = rs;
                        init = true;
                    }
                } finally {
                    cellsBusy = 0;
                }
                if (init)
                    break;
            }
            else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
                break;                          // Fall back on using base
        }
    }

流程大概为:

image.png

小结:
1, 元素数量的维护主要依靠 baseCount 和 countCells数组完成
2, 如果不能在baseCount中添加数量, 则再countCells数组中添加
3, 根据变量 cellBusy的CAS操作来控制线程安全性, 为0时方可进行操作
4, 数组的初始长度为2, 每次扩容为原来的两倍, 长度达到CPU最大核数后不扩容.

八, 扩容

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            // 最小划分模块 16 个
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        if (nextTab == null) {            // initiating
            // 创建新的数组, 长度为原来的2倍
            try {
                @SuppressWarnings("unchecked")
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
                nextTab = nt;
            } catch (Throwable ex) {      // try to cope with OOME
                sizeCtl = Integer.MAX_VALUE;
                return;
            }
            nextTable = nextTab;
            transferIndex = n;
        }
        int nextn = nextTab.length;
        // 迁移节点, hash值为MOVED(-1), put方法里有用
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        boolean advance = true;
        boolean finishing = false; // to ensure sweep before committing nextTab
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            // 1,分配迁移的坐标, 涉及到多线程的任务分配
            while (advance) {
                int nextIndex, nextBound;
               // 每个线程执行完第一次迁移之后, 每次会将索引-1, 判断是否超过上次分配的索引
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                // 第一次进入while和第一次分配任务完成后进入, 重新分配任务
                else if (U.compareAndSwapInt
                         (this, TRANSFERINDEX, nextIndex,
                          nextBound = (nextIndex > stride ?
                                       nextIndex - stride : 0))) {
                    // 索引访问
                    bound = nextBound;
                    // 开始索引
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            // 2, 完成迁移后退出
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                if (finishing) {
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    finishing = advance = true;
                    i = n; // recheck before commit
                }
            }
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                // 3, 迁移
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        if (fh >= 0) {
                            -- 普通节点迁移 --
                        }
                        else if (f instanceof TreeBin) {
                           -- 树节点迁移 --
                        }
                    }
                }
            }
        }
    }

小结:
1, 多线程如何协助迁移. 当在扩容时, 完成迁移扩容的单元格哈希值为-1, 此时线程不执行添加, 先执行扩容任务. 扩容任务执行时, 将旧数组分为N(N>=16)段, 每次任务(可能是多个线程, 也可能是单个线程把前面的任务执行完了, 再回来拿任务)划分一段, 从后往前迁移. 通过全局变量transferIndex控制.

n, 总结

1, ConcurrentHashMap很少使用锁, 大部分使用CAS+自旋的方式,保证线程安全.
2, 默认容量是16, 扩容阀值为当前容量*3/4, 扩容为原容量的2倍, 加载因子不影响阀值计算.
3, 由链表转为树结构的条件: 链表的长度达到8,且数组长度达到64, 如果数组长度小数64 , 则进行扩容. (数组可能有些值还没有填充)
4, 元素数量的维护主要是 依靠 baseCount 和 countCells数组完成.
5, 多线程如何协助迁移. 当在扩容时, 完成迁移扩容的单元格哈希值为-1, 此时线程不执行添加, 先执行扩容任务. 扩容任务执行时, 将旧数组分为N(N>=16)段, 每次任务(可能是多个线程, 也可能是单个线程把前面的任务执行完了, 再回来拿任务)划分一段, 从后往前迁移. 通过全局变量transferIndex控制.