# ConcurrentHashMap

# 并发特点

死循环处理修改，处理不成功则更新到最新状态，处理成功则跳出

# 一、基础

数组 + 链表 + 红黑树

  1. 链表和红黑树会相互转换
  2. 多线程按分配步长进行扩容任务分配
  3. 数组戳记使用
  4. 哈希混淆
  5. 头节点分类
  6. 分格计数
private static final int MAXIMUM_CAPACITY = 1 << 30;
private static final int MIN_TRANSFER_STRIDE = 16;
private static 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;

# 1.1 节点类型：

  1. 链表节点： Node
  2. 红黑树：TreeBin + TreeNode
  3. 扩容节点： ForwardingNode
  4. 占位节点：ReservationNode
static final int MOVED     = -1;
static final int TREEBIN   = -2; // hash for roots of trees
static final int RESERVED  = -3; //computeIfAbsent and compute

# ForwardingNode作用

  1. 扩容时的头节点，标志某个桶已经被复制到了新的桶数组中,
  2. 转发get操作，避免阻塞get方法的调用
  3. 协助扩容

# 1.2 属性：

  1. 容量最大值2^30
private static final int MAXIMUM_CAPACITY = 1 << 30;
  1. 控制符
# 当为负数时：-1代表正在初始化，-N代表有N-1个线程正在 进行扩容
# 当为0时：代表当时的table还没有被初始化
# 当为正数时：表示初始化或者下一次进行扩容的大小
private transient volatile int sizeCtl;

# 数组未初始化时，sizeCtl=初始容量。
# 数组正在初始化时，sizeCtl=-1。
# 数组初始化完成，sizeCtl=扩容阈值。
# 数组扩容时，sizeCtl被赋值一个非常小的负数，控制扩容线程数量的加减以及用来标识数组正在扩容的状态。
  • 线程通过CAS将sizeCtl从0变为-1，则获取锁成功，可以初始化table
  • 其他线程检测到sizeCtl小于0，则Thread.yield()
  1. 扩容位置

transferIndex，记录了扩容任务分配的进度。初始为n，逆序扩容，每次减一个步长的位置，最终减至＜=0，表示整个扩容任务分配完了。

# 数组stamp

static final int resizeStamp(int n) {
    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

// RESIZE_STAMP_BITS = 16

数组stamp：随数组长度变化，长度一定，stamp一致。高16为0，使用sizeCtl高16为保存。

这个方法的返回是在2^16 - 2^16+32之间

# 二、主要实现：

# 0. 节点

# 1. 链表节点

Node<K,V> find(int h, Object k) {
    Node<K,V> e = this;
    if (k != null) {
        do {
            K ek;
            if (e.hash == h &&
                ((ek = e.key) == k || (ek != null && k.equals(ek))))
                return e;
        } while ((e = e.next) != null);
    }
    return null;
}

# 2. 树节点

红黑树的节点有两种，

  • TreeBin是根节点，是一个空节点，负责红黑树添加、删除、查找节点
  • TreeNode是真正存key-value的节点。

TreeBin维护一个读写锁，目的是在新增和删除节点时，维护红黑树的结构平衡过程进行加锁。新节点以头插法的方式串成链表，所以修改不会影响遍历过程且next指针被volatile修饰，修改指针后会立即通知到所有线程获取最新值。

读写锁使用lockState记录锁的状态，有三种标志位：

  • WRITER=1，二进制低位第一位用来标识写线程持有锁的状态 （不可重入写锁）。
  • WAITER=2，二进制低位第二位用来标识阻塞状态。
  • WAITER=4，二进制低位第三位之后都是用来标识读线程持有锁的状态。
static final int WRITER = 1; // set while holding write lock
static final int WAITER = 2; // set when waiting for write lock
static final int READER = 4; // increment value for setting read lock

读读不互斥，lockState+READER代表一个读线程获取锁，lockState-READER代表一个读线程释放锁。

# 2.1 TreeBin

// todo

# 2.2 TreeNode

负责节点查找

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;

    TreeNode(int hash, K key, V val, Node<K,V> next,
             TreeNode<K,V> parent) {
        super(hash, key, val, next);
        this.parent = parent;
    }

    Node<K,V> find(int h, Object k) {
        return findTreeNode(h, k, null);
    }

    /**
     * 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) {
        if (k != null) {
            TreeNode<K,V> p = this;
            do  {
                int ph, dir; K pk; TreeNode<K,V> q;
                TreeNode<K,V> pl = p.left, pr = p.right;
                if ((ph = p.hash) > h)
                    p = pl;
                else if (ph < h)
                    p = pr;
                else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
                    return p;
                else if (pl == null)
                    p = pr;
                else if (pr == null)
                    p = pl;
                else if ((kc != null ||
                          (kc = comparableClassFor(k)) != null) &&
                         (dir = compareComparables(kc, k, pk)) != 0)
                    p = (dir < 0) ? pl : pr;
                else if ((q = pr.findTreeNode(h, k, kc)) != null)
                    return q;
                else
                    p = pl;
            } while (p != null);
        }
        return null;
    }
}

# 3. 扩容节点

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);
        this.nextTable = tab;
    }
    
    Node<K,V> find(int h, Object k) {
        // loop to avoid arbitrarily deep recursion on forwarding nodes
        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) { //还是扩容节点
                        tab = ((ForwardingNode<K,V>)e).nextTable;
                        continue outer;
                    }
                    else
                        return e.find(h, k);
                }
                if ((e = e.next) == null)
                    return null;
            }
        }
    }
}

# 1. 初始化

try {
    if (table是为空) {
        初始化table
        sc = n - (n >>> 2);
    }
} finally {
    sizeCtl = sc;
}

sizeCtl保存需要扩容的大小

# 2. 并发扩容：

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        int rs = resizeStamp(tab.length);
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}
  • (sc >>> RESIZE_STAMP_SHIFT) != rs 保证所有线程要基于同一个旧的桶数组扩容
  • transferIndex <= 0 已经有线程完成扩容任务了
  • sc == (rs << RESIZE_STAMP_SHIFT) + 1当前扩容的线程数为0，即已经扩容完成了，就不需要再新增线程扩容
  • sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS参与扩容的线程数已经到了最大，就不需要再新增线程扩容

至于sc == rs + 1 || sc == rs + MAX_RESIZERS这两个判断条件是JDK的一个BUG，这个BUG已经在JDK 12中修复，可以参考一下Oracle的官网：https://bugs.java.com/bugdatabase/view_bug.do?bug_id=JDK-8214427，这两个判断条件应该写成这样：sc == (rs << RESIZE_STAMP_SHIFT) + 1 || sc == (rs << RESIZE_STAMP_SHIFT) + MAX_RESIZERS,因为直接比较rs和sc是没有意义的，必须要有移位操作。

实际扩容时，从n->2n的过程中，每个线程申请一定的步长进行扩容。

(sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT

说明是最后一个扩容线程，进行最后的处理。

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)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    if (nextTab == null) {            // initiating
        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;
    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;
        ... 计算扩容范围
        if (i < 0 || i >= n || i + n >= nextn) {
            ... 退出条件
        }
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            ... 移动元素
        }
    }
}
  1. 计算布长
  2. 初始化nexttab，更新全局nextTab和transferIndex
  3. 计算扩容范围，进行移动元素，在满足条件时退出

循环

# 范围计算
while (advance) {
    int nextIndex, nextBound;
    if (--i >= bound || finishing)
        advance = false;
    else if ((nextIndex = transferIndex) <= 0) {
        i = -1;
        advance = false;
    }
    else if (U.compareAndSwapInt
             (this, TRANSFERINDEX, nextIndex,
              nextBound = (nextIndex > stride ?
                           nextIndex - stride : 0))) {
        bound = nextBound;
        i = nextIndex - 1;
        advance = false;
    }
}

获得处理范围[transferIndex-stride, transferIndex-1]

  • 左边界bound = transferIndex-stride
  • 右边界i = nextIndex - 1
# 退出条件
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
    }
}
  1. 扩容完成finishing=ture
  2. 数组的stamp变了，数组大小右一次变化，（sc - 2）表示没有线程在扩容
# 移动
synchronized (f) {
    if (tabAt(tab, i) == f) {
        Node<K,V> ln, hn;
        if (fh >= 0) {
            int runBit = fh & n;
            Node<K,V> lastRun = f;
            for (Node<K,V> p = f.next; p != null; p = p.next) {
                int b = p.hash & n;
                if (b != runBit) {
                    runBit = b;
                    lastRun = p;
                }
            }
            if (runBit == 0) {
                ln = lastRun;
                hn = null;
            }
            else {
                hn = lastRun;
                ln = null;
            }
            for (Node<K,V> p = f; p != lastRun; p = p.next) {
                int ph = p.hash; K pk = p.key; V pv = p.val;
                if ((ph & n) == 0)
                    ln = new Node<K,V>(ph, pk, pv, ln);
                else
                    hn = new Node<K,V>(ph, pk, pv, hn);
            }
            setTabAt(nextTab, i, ln);
            setTabAt(nextTab, i + n, hn);
            setTabAt(tab, i, fwd);
            advance = true;
        }
        else if (f instanceof TreeBin) {
            TreeBin<K,V> t = (TreeBin<K,V>)f;
            TreeNode<K,V> lo = null, loTail = null;
            TreeNode<K,V> hi = null, hiTail = null;
            int lc = 0, hc = 0;
            for (Node<K,V> e = t.first; e != null; e = e.next) {
                int h = e.hash;
                TreeNode<K,V> p = new TreeNode<K,V>
                    (h, e.key, e.val, null, null);
                if ((h & n) == 0) {
                    if ((p.prev = loTail) == null)
                        lo = p;
                    else
                        loTail.next = p;
                    loTail = p;
                    ++lc;
                }
                else {
                    if ((p.prev = hiTail) == null)
                        hi = p;
                    else
                        hiTail.next = p;
                    hiTail = p;
                    ++hc;
                }
            }
            ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                (hc != 0) ? new TreeBin<K,V>(lo) : t;
            hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                (lc != 0) ? new TreeBin<K,V>(hi) : t;
            setTabAt(nextTab, i, ln);
            setTabAt(nextTab, i + n, hn);
            setTabAt(tab, i, fwd);
            advance = true;
        }
    }
}
  1. 双重检查加锁时，头节点没有改变
  2. 链表和树节点的转移

# 3. put：

for循环中初始化中多次更新条件，最后通过synchronized对头节点加锁进行更新。

加锁成功再次判断，防止扩容线程导致头节点已经变化

tabAt(tab, i) == f

# 4. count：

设计思想：分布式计数(distributing counts)，通过baseCount与counterCells数组组合.

总数就是baseCount和每个cell里的值的和。

Adapted from LongAdder and Striped64

private transient volatile long baseCount;
private transient volatile CounterCell[] counterCells;

//伪共享问题
@sun.misc.Contended static final class CounterCell {
    volatile long value;
    CounterCell(long x) { value = x; }
}
  • 当没有线程竞争时，通过cas增加baseCount即可
  • 当竞争失败，通过不同的cell将竞争拆分，称为分布式计数。
private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        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) {
        ...
    }
}

第一个if用来添加计数

  • 如果counterCells已经初始化，则不尝试cas，直接按分散到cell进行计数
  • 数组没有初始化，对应cell没有初始化，对cell进行cas失败，则进入fullAddCount

第二个if用来检查是否需要扩容。

# getProbe() 线程的身份
static final int getProbe() {
    return UNSAFE.getInt(Thread.currentThread(), PROBE);
}

getProbe方法会返回当前线程的一个唯一身份码，这个值是不会变的，它的返回值可能是0，如果返回0则需要调用ThreadLocalRandom.localInit()初始化。

# fullAddCount 一定要加成功
private final void fullAddCount(long x, boolean wasUncontended) {
    int h;
    if ((h = ThreadLocalRandom.getProbe()) == 0) {
        ThreadLocalRandom.localInit();      // force initialization
        h = ThreadLocalRandom.getProbe();
        wasUncontended = 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) {
            if ((a = as[(n - 1) & h]) == null) {
                if (cellsBusy == 0) {            // Try to attach new Cell
                    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;
                        }
                        if (created)
                            break;
                        continue;           // Slot is now non-empty
                    }
                }
                collide = false;
            }
            else if (!wasUncontended)       // CAS already known to fail
                wasUncontended = true;      // Continue after rehash
            else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
                break;
            else if (counterCells != as || n >= NCPU)
                collide = false;            // At max size or stale
            else if (!collide)
                collide = true;
            else if (cellsBusy == 0 &&
                     U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                try {
                    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)) {
            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
    }
}

进入方法的情况

  1. CounterCell是null
  2. CounterCell长度是0
  3. CounterCell对应probe位置是null
  4. CounterCell对应probe位置cas失败
  5. probe的值返回0，需要重新计算

cellsBusy作为一个锁，用来控制CounterCell初始化，扩容，cell初始化

并发添加计数需要处理上面情况，并考虑并发情况

  • 首先处理probe是否是0导致的并发竞争，并设置为不是竞争（wasUncontended）
for (;;) {
    CounterCell[] as; CounterCell a; int n; long v;
    if ((as = counterCells) != null && (n = as.length) > 0) {
        ...
    }
    else if (cellsBusy == 0 && counterCells == as &&
             U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        ...
    }
    else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
        break;                          // Fall back on using base
}

外层IF，成功则break，否则继续循环

  • counterCells已经初始化，只需要定位cell，并cas计算
  • counterCells未初始化，加锁cellsBusy，并初始化
  • 再试下basecount

内层IF，counterCells已经初始化

  • 对应的位置没有初始化，则加锁，初始化
  • 若已知有竞争，则下一轮，和spin功能一样
  • cas，成功则退出，否则下一轮
  • 标记为cas失败，有冲突，下一轮进行扩容
  • 加锁，扩容
  • 更新probe，重新开始

# 树

# treeifyBin
private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    if (tab != null) {
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            tryPresize(n << 1);
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            synchronized (b) {
                if (tabAt(tab, index) == b) {
                    TreeNode<K,V> hd = null, tl = null;
                    for (Node<K,V> e = b; e != null; e = e.next) {
                        TreeNode<K,V> p =
                            new TreeNode<K,V>(e.hash, e.key, e.val,
                                              null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    setTabAt(tab, index, new TreeBin<K,V>(hd));
                }
            }
        }
    }
}

TreeNode形成的只是链表，通过将头节点初始TreeBin才能构造红黑树

并发特点
一、基础
二、主要实现：
0. 节点
1. 初始化
2. 并发扩容：
3. put：
4. count：