JDK 源码解析 —— ConcurrentHashMap

类图： 

抽象结构图：

// 在准备锁住 segment 操作前最大的 tryLock() 次数。 多核情况下， 在定位 nodes 时使用 64 次最大值维持缓存

static final int MAX_SCAN_RETRIES =
    Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;

/**
 * The per-segment table. 数据访问通过
 * entryAt/setEntryAt 提供的 volatile 语义来保证可见性.
 */
transient volatile HashEntry<K,V>[] table;

/**
 * sengment 内 hash entry 元素个数 , 之所以在每个 Segment 对象中包含一个计数器，而不是在 ConcurrentHashMap 中使用全局的计数器，是为了避免出现“热点域”而影响 ConcurrentHashMap 的并发性。

 */
transient int count;

/**
 * 在 segment 中可变操作总数, 即更新次数
 */
transient int modCount;

/**
 * 当超过threshold 时候，  table 再哈希
 * (The value of this field is always <tt>(int)(capacity *
 * loadFactor)</tt>.)
 */
transient int threshold;

/**
 *  hash table 负载因子.  Even though this value
 * is same for all segments, it is replicated to avoid needing
 * links to outer object.
 * @serial
 */
final float loadFactor;

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
 // tryLock 一般缓存作用
HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        HashEntry<K,V>[] tab = table;
        int index = (tab.length - 1) & hash;
        // 找到 bucket 位置
        HashEntry<K,V> first = entryAt(tab, index);
        for (HashEntry<K,V> e = first;;) {
            if (e != null) {
                K k;
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) {
                    // 保存旧值， 这样 get 操作就可以无锁访问正在写操作的节点
                    oldValue = e.value;
                    if (!onlyIfAbsent) {
                        // 覆盖原来的值
                        e.value = value;
                        // 修改计数
                        ++modCount;
                    }
                    break;
                }
                // 每次从头部插入
                e = e.next;
            }
            else {
                if (node != null)
                    node.setNext(first);
                else
                    node = new HashEntry<K,V>(hash, key, value, first);
                int c = count + 1;
                // 超过 threshold 则， rehash()
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    rehash(node);
                else
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        // 典型的 ReentrantLock 释放锁
        unlock();
    }
    return oldValue;
}

rehash() 方法:  在新 table 中重新分类节点。 因为使用了 2 的幂指数扩展方式， bucket/bin 中的数据还是在原位， 即旧数据的索引位置不变或者偏移了 2 的幂指数距离。 可以重用旧节点减少不必要的节点生成。 

/**
 * table 大小 *2， 重新放置 HashEntry， 加入新节点
 */
@SuppressWarnings("unchecked")
private void rehash(HashEntry<K,V> node) {
    // 保存旧值以便 get 操作遍历
    HashEntry<K,V>[] oldTable = table;
    int oldCapacity = oldTable.length;
    int newCapacity = oldCapacity << 1;
    threshold = (int)(newCapacity * loadFactor);
    HashEntry<K,V>[] newTable =
        (HashEntry<K,V>[]) new HashEntry[newCapacity];
    int sizeMask = newCapacity - 1;
    for (int i = 0; i < oldCapacity ; i++) {
        HashEntry<K,V> e = oldTable[i];
        if (e != null) {
            HashEntry<K,V> next = e.next;
            int idx = e.hash & sizeMask;
            if (next == null)   //  单节点链表
                newTable[idx] = e;
            else { // 重用在同一个 slot 的连续序列
                HashEntry<K,V> lastRun = e;
                int lastIdx = idx;
                for (HashEntry<K,V> last = next;
                     last != null;
                     last = last.next) {
                    int k = last.hash & sizeMask;
                    if (k != lastIdx) {
                        lastIdx = k;
                        lastRun = last;
                    }
                }
                newTable[lastIdx] = lastRun;
                // Clone remaining nodes
                for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
                    V v = p.value;
                    int h = p.hash;
                    int k = h & sizeMask;
                    HashEntry<K,V> n = newTable[k];
                    newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
                }
            }
        }
    }
    int nodeIndex = node.hash & sizeMask; // add the new node
    node.setNext(newTable[nodeIndex]);
    newTable[nodeIndex] = node;
    table = newTable;
}

内部类 HashEntry：链表结构

static final class HashEntry<K,V> {
    // final 保证不变性， 即为线程安全的字段
    final int hash;
    final K key;
    // 根据 volatile 写永远先于读操作的 happens-before 原则来保证获取到都是最新值
    volatile V value;
    volatile HashEntry<K,V> next;

    HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.value = value;
        this.next = next;
    }

    /**
     * 使用 volatile 写语义设置 next
     */
    final void setNext(HashEntry<K,V> n) {
        UNSAFE.putOrderedObject(this, nextOffset, n);
    }

    // Unsafe mechanics
    static final sun.misc.Unsafe UNSAFE;
    static final long nextOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class k = HashEntry.class;
            nextOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("next"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }
}

ConcurrentHashMap 类：

/* ---------------- Constants -------------- */

/**
 * The default initial capacity for this table,
 * used when not otherwise specified in a constructor.
 */
static final int DEFAULT_INITIAL_CAPACITY = 16;

/**
 * 本值是 HashEntry 个数与 table 数组长度的比值
 *
 */
static final float DEFAULT_LOAD_FACTOR = 0.75f;

/**
 *
 * 当前并发线程的使用数
 */
static final int DEFAULT_CONCURRENCY_LEVEL = 16;

/**
 * The maximum capacity, used if a higher value is implicitly
 * specified by either of the constructors with arguments.  MUST
 * be a power of two <= 1<<30 to ensure that entries are indexable
 * using ints.
 */
static final int MAXIMUM_CAPACITY = 1 << 30;

/**
 * The minimum capacity for per-segment tables.  Must be a power
 * of two, at least two to avoid immediate resizing on next use
 * after lazy construction.
 */
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;

/**
 * The maximum number of segments to allow; used to bound
 * constructor arguments. Must be power of two less than 1 << 24.
 */
static final int MAX_SEGMENTS = 1 << 16; // slightly conservative

/**
 * Number of unsynchronized retries in size and containsValue
 * methods before resorting to locking. This is used to avoid
 * unbounded retries if tables undergo continuous modification
 * which would make it impossible to obtain an accurate result.
 */
static final int RETRIES_BEFORE_LOCK = 2;
/**
 * 索引 segments 时使用： 使用高比特位的 hash 值去选择 segment
 */
final int segmentMask;

/**
 * Shift value for indexing within segments.
 */
final int segmentShift;

ConcurrentHashMap 结构图

public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    if (concurrencyLevel > MAX_SEGMENTS)
        concurrencyLevel = MAX_SEGMENTS;
    // Find power-of-two sizes best matching arguments
    int sshift = 0;
    int ssize = 1;
    while (ssize < concurrencyLevel) {
        ++sshift;
        ssize <<= 1;
    }
    this.segmentShift = 32 - sshift;
    this.segmentMask = ssize - 1;
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c)
        cap <<= 1;
    // create segments and segments[0]
    Segment<K,V> s0 =
        new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
                         (HashEntry<K,V>[])new HashEntry[cap]);
    Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

JDK 源码解析 —— ConcurrentHashMap

原文：http://blog.csdn.net/wenniuwuren/article/details/49108341