JDK1.7 HashMap

HashMap的初始化(构造函数)

1、不指定大小

如果不指定大小，那么HashMap默认大小为16，加载因子是0.75

public HashMap() 
{this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR);  //如果不指定大小，那么HashMap默认大小为16，加载因子是0.75
}

2、指定大小

如果初始化容量大于最大容量1<<30，则设置成最大容量，

public HashMap(int initialCapacity, float loadFactor) 
{if (initialCapacity < 0)throw new IllegalArgumentException("Illegal initial capacity: " +initialCapacity); if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;if (loadFactor <= 0 || Float.isNaN(loadFactor))throw new IllegalArgumentException("Illegal load factor: " +loadFactor);this.loadFactor = loadFactor;threshold = initialCapacity;init();  //空函数
}

put

public V put(K key, V value) 
{if (table == EMPTY_TABLE) {//1、将HashMap大小修改成2的n次方//2、计算阈值即容量*加载因子//3、创建大小为capacity的Entry数组，Entry中存放了hash值，key，value等数据//4、调用initHashSeedAsNeeded函数 	inflateTable(threshold); }if (key == null)//1、一定要注意桶0里面不是都是null的，还存放了通过按位与计算出来正好为0的key value。//2、如果桶0有key为null的数据，则替换掉。//3、如果没有调用addEntry(0, null, value, 0);//参数分别是hash、key、value、bucketIndex//4、如果addEntry的时候发现size大于等于阈值，而且Entry[bucketIndex]!=null，则进行resize扩容，对元素进行rehash操作，然后存放元素，如果不需要扩容，就直接使用头插法存放return putForNullKey(value);//相当于计算hash值int hash = hash(key);//计算元素应该存在哪个桶int i = indexFor(hash, table.length);//遍历元素应该存放的桶，找有没有hash相同且key和存放元素相同的元素，如果有直接修改，没有的话插入元素，插入元素的时候还是要判断是否需要扩容需要进行rehash，再对元素进行插入。for (Entry<K,V> e = table[i]; e != null; e = e.next) {Object k;if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {V oldValue = e.value;e.value = value;e.recordAccess(this);return oldValue;}}modCount++;addEntry(hash, key, value, i);return null;
}

get

public V get(Object key) 
{if (key == null)//	如果key为null，遍历桶0return getForNullKey();Entry<K,V> entry = getEntry(key);//不是null的话，先计算key的hash值，计算应该存在哪个桶，再去那个桶中去找。看是否有hash值相同且key相同的元素return null == entry ? null : entry.getValue();
}

size

public int size() {return size;
}

HashMap线程不安全体现

1.7 HashMap扩容时候的死锁问题

阅读HashMap源码的时候我们知道，HashMap在扩容的时候会调用transfer函数。就是将原Hash表上的元素全部转移到新的hash表上，这里贴出transfer的函数：

void transfer(Entry[] newTable, boolean rehash) 
{int newCapacity = newTable.length;for (Entry<K,V> e : table) //遍历所有的hash桶{while(null != e) //遍历每个桶下面的链表{Entry<K,V> next = e.next; if (rehash) {e.hash = null == e.key ? 0 : hash(e.key);}int i = indexFor(e.hash, newCapacity);e.next = newTable[i]; //下面两行使用头插入法插入元素newTable[i] = e;e = next; //原来的链表遍历到下一个节点}}
}

正常模式下的rehash，一个个通过头插入法插入到新的哈希桶中。

并发下的rehash

void transfer(Entry[] newTable, boolean rehash) 
{int newCapacity = newTable.length;for (Entry<K,V> e : table) //遍历所有的hash桶{while(null != e) //遍历每个桶下面的链表{Entry<K,V> next = e.next;  //if (rehash) {e.hash = null == e.key ? 0 : hash(e.key);}int i = indexFor(e.hash, newCapacity);e.next = newTable[i]; //下面两行使用头插入法插入元素newTable[i] = e;e = next; //原来的链表遍历到下一个节点}}
}

如果线程一执行到Entry<K,V> next = e.next;被挂起。而线程二执行完成了，就会出现下面的样子。

注意，因为Thread1的 e 指向了key(3)，而next指向了key(7)，其在线程二rehash后，指向了线程二重组后的链表。我们可以看到链表的顺序被反转后。

线程一如果被调度回来，会执行下面的代码：

• e.next = newTable[i]; //下面两行使用头插入法插入元素

• 线程一被调度回来会执行newTable[i] = e;

• 然后e = next; 导致e指向了key(7)

• 下一次循环的next = e.next；导致了next指向了key(3)。

线程一接着工作。把key(7)摘下来，放到newTable[i]的第一个，然后把e和next往下移。

环形链接出现。
e.next = newTable[i] 导致 key(3).next 指向了 key(7)

注意：此时的key(7).next 已经指向了key(3)， 环形链表就这样出现了。

1.8 HashMap

创建的是Node数组。

构造函数

如果初始化容量大于最大容量1<<30，则设置成最大容量，

public HashMap(int initialCapacity, float loadFactor) 
{if (initialCapacity < 0)throw new IllegalArgumentException("Illegal initial capacity: " +initialCapacity);if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;if (loadFactor <= 0 || Float.isNaN(loadFactor))throw new IllegalArgumentException("Illegal load factor: " +loadFactor);this.loadFactor = loadFactor;//这行与1.7不同，1.7在put中进行做。//获得大于capacity的最小的2的幂  this.threshold = tableSizeFor(initialCapacity);
}

put函数

put函数

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

hash函数

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

putVal函数

final V putVal(int hash, K key, V value, boolean onlyIfAbsent,boolean evict) 
{Node<K,V>[] tab; Node<K,V> p; int n, i;//第一次put元素时，table数组为空，先调用resize生成一个指定容量的数组if ((tab = table) == null || (n = tab.length) == 0)n = (tab = resize()).length;if ((p = tab[i = (n - 1) & hash]) == null)//如果计算出来的那个桶里面没有元素则直接存放tab[i] = newNode(hash, key, value, null);else {Node<K,V> e; K k;//在if那p被设置成了通过按位与计算出来的第一个元素//如果hash值和插入的key的hash值相同且key相同if (p.hash == hash &&((k = p.key) == key || (key != null && key.equals(k))))//如果和桶的第一个元素相等，则直接替换e = p;  //判断当前Node是否是红黑树的节点else if (p instanceof TreeNode)e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);else {for (int binCount = 0; ; ++binCount) {//使用尾插法插入元素if ((e = p.next) == null) {p.next = newNode(hash, key, value, null);//当链表长度大于等于8，加上桶里面那个就应该是大于8。if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1sttreeifyBin(tab, hash);break;}//如果插入的key和这个桶中某元素的key相等且元素的hash和插入的key计算出来的hash相等则break，在下面进行修改。if (e.hash == hash &&((k = e.key) == key || (key != null && key.equals(k))))break;p = e;}}if (e != null) { // existing mapping for keyV oldValue = e.value;if (!onlyIfAbsent || oldValue == null)e.value = value;afterNodeAccess(e);return oldValue;}}++modCount;//如果当前size超过临界值，就扩容。注意是先插入节点再扩容if (++size > threshold)resize();afterNodeInsertion(evict);return null;
}

treeifyBin

final void treeifyBin(Node<K,V>[] tab, int hash) 
{int n, index; Node<K,V> e;//链表长度超过8且数组容量超过64时，将链表转换为红黑树if (tab == null || (n = tab.length) < MIN_TREEIFY_CAPACITY)resize();else if ((e = tab[index = (n - 1) & hash]) != null) {TreeNode<K,V> hd = null, tl = null;do {TreeNode<K,V> p = replacementTreeNode(e, null);if (tl == null)hd = p;else {p.prev = tl;tl.next = p;}tl = p;} while ((e = e.next) != null);if ((tab[index] = hd) != null)hd.treeify(tab);}
}

在1.7中是插入的时候就看需不需要扩容，而1.8是先插入进去再扩容，1.7使用的是头插法，1.8使用的是尾插法。

加载因子为什么是0.75

加载因子为什么是0.75

根据泊松分布计算而来，根据泊松分布计算可以得到在加载因子是0.75的时候，链表长度大于8几乎是不可能的。

泊松分布描述的是某段时间内，时间具体发生的概率。

• t表示时间
• n表示数量
• λ 表示事件的频率

平均每小时生三个婴儿，下面计算的是接下来两个小时，一个婴儿都不出生的概率是0.25%.基本不可能发生。

ConcurrentHashMap

参考链接

Hashtable的效率低下因为对put等操作使用了synchronized关键字进行加锁，也就是说在进行put等修改Hash表的操作的时候，锁住了整个Hash表，从而使得整个Hash表的效率低下。

JDK1.7之前的ConcurrentHashMap使用分段锁机制实现，JDK1.8则使用数组+链表+红黑树数据结构和CAS原子操作实现ConcurrentHashMap；本文将分别介绍这两种方式的实现方案及其区别。

1.7(并发度会收到Segment的个数的限制)

1、分段锁机制

ConcurrentHashMap中包含Segment数组，每个Segment中又包含HashEntry数组，Segment继承自ReentrantLock

在JDK1.7中，Java使用分段锁机制实现ConcurrentHashMap，在ConcurrentHashMap对象中有一个Segment数组(Segment类继承ReentrantLock类)，即将整个Hash表划分成了多个Segment分段，即每个分段则类似于一个Hashtable，这样在执行put操作的时候首先根据hash算法定位到属于哪个Segment，然后对Segment加锁即可。因此ConcurrentHashMap在多线程并发执行的过程中实现了多线程put操作。

ConcurrentHashMap类结构如上图所示。由图可知，在ConcurrentHashMap中，定义了一个Segment<K, V>[]数组来将Hash表实现分段存储，从而实现分段加锁；而么一个Segment元素则与HashMap结构类似，其包含了一个HashEntry数组，用来存储Key/Value对。Segment继承了ReetrantLock，表示Segment是一个可重入锁，因此ConcurrentHashMap通过可重入锁对每个分段进行加锁。

2、ConcurrentHashMap的初始化

ConcurrentHashMap包含多个构造函数，而所有的构造函数最终都调用了如下的构造函数：

@SuppressWarnings("unchecked")
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 argumentsint sshift = 0;//Segment的大小，分段数组的大小，如果小于concurrencyLevel//会被调整成大于或等于concurrencyLevel的最小的2的N次方值来作为segments数组的长度//2^sshift=ssize  int ssize = 1; while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}//一个键值对在Segment数组中下标j的计算公式为：//比如ssize=16,则segmentMask为1111b，//j = (hash >>> segmentShift) & segmentMask//2^sshift=ssize  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;
}

• ssize表示了segments数组的长度(为了能通过按位与的散列算法来定位segments数组的索引，必须保证segments数组的长度是2的N次方)
• sshift表示了计算ssize时进行移位操作的次数。
• segmentShift用于定位参与散列运算的位数，其等于32减去sshift，使用32是因为ConcurrentHashMap的hash()方法返回的最大数是32位的,hash >>> segmentShift所以就会只剩下低sshift位为1，其余都为0，因此，求key散列到长度为ssize的Segment数组的下标j，就是求key的hash值的高sshift位。

总的来说，计算在Segment的哪个位置就是计算Segment的高sshift位与segmentMask进行按位与得到元素位于位于Segment中的下标。

由此可知，求key散列到长度为ssize的Segment数组的下标j，必定有下标j的值域为[0，ssize-1]。由于ssize=2^sshift，那么小标j可以用1个sshift位的二进制数字表示。假如：ssize为16，那么sshift=4，j的值域为[0,15]，而[0000b,1111b]就是j的值域；则求key在Segment[]的下标j，就是求key对应的一个散列的4位二进制数值。而ConcurrentHashMap的源码求下标j的方式非常简单，就是取key的hash值的高4位。因此，求key散列到长度为ssize的Segment数组的下标j，就是求key的hash值的高sshift位。

由代码可知，该构造函数需要传入三个参数：initialCapacity、loadFactor、concurrencyLevel，其中，concurrencyLevel主要用来初始化segments、segmentShift和segmentMask等；而initialCapacity和loadFactor则主要用来初始化每个Segment分段。

ConcurrentHashMap的get操作(不加锁)

JDK1.7的ConcurrentHashMap的get操作是不加锁的，因为在每个Segment中定义的HashEntry数组和在每个HashEntry中定义的value和next HashEntry节点都是volatile类型的，volatile类型的变量可以保证其在多线程之间的可见性，因此可以被多个线程同时读，从而不用加锁。而其get操作步骤也比较简单，定位Segment –> 定位HashEntry –> 通过getObjectVolatile()方法获取指定偏移量上的HashEntry –> 通过循环遍历链表获取对应值。

static final class HashEntry<K,V> {final int hash;final K key;volatile V value;volatile HashEntry<K,V> next;

public V get(Object key) 
{Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;int h = hash(key);long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);e != null; e = e.next) {K k;if ((k = e.key) == key || (e.hash == h && key.equals(k)))return e.value;}}return null;
}

• 定位Segment：(((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE

• 定位HashEntry：(((tab.length - 1) & h)) << TSHIFT) + TBASE

3、ConcurrentHashMap的put函数

public V put(K key, V value) 
{Segment<K,V> s;if (value == null)throw new NullPointerException();int hash = hash(key);//计算元素应该放在哪个Segment数组的哪个位置int j = (hash >>> segmentShift) & segmentMask;//SBASE = UNSAFE.arrayBaseOffset(sc);数组的起始地址//下面就相当于拿到元素存放的Segment数组中的那个Segment元素if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck(segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegments = ensureSegment(j);return s.put(key, hash, value, false);
}

ensureSegment函数

private Segment<K,V> ensureSegment(int k) 
{final Segment<K,V>[] ss = this.segments;long u = (k << SSHIFT) + SBASE; // raw offset，物理地址Segment<K,V> seg;//在put函数中一开始调用的是getObject函数，后来在这里调用的是getObjectVolatile函数if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {Segment<K,V> proto = ss[0]; // use segment 0 as prototypeint cap = proto.table.length;float lf = proto.loadFactor;int threshold = (int)(cap * lf);HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) { // recheckSegment<K,V> s = new Segment<K,V>(lf, threshold, tab);while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) {if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))break;}}}return seg;
}

同样的，put方法首先也会通过hash算法定位到对应的Segment，此时，如果获取到的Segment为空，则调用ensureSegment()方法；否则，直接调用查询到的Segment的put方法插入值，注意此处并没有用getObjectVolatile()方法读，而是在ensureSegment()中再用volatile读操作，这样可以在查询segments不为空的时候避免使用volatile读，提高效率。在ensureSegment()方法中，首先使用getObjectVolatile()读取对应Segment，如果还是为空，则以segments[0]为原型创建一个Segment对象，并将这个对象设置为对应的Segment值并返回，通过上面的CAS进行设置。

在Segment的put方法中，首先需要调用tryLock()方法获取锁，然后通过hash算法定位到对应的HashEntry，然后遍历整个链表，如果查到key值，则直接插入元素即可；而如果没有查询到对应的key且个数大于域值，则需要调用rehash()方法对Segment中保存的table进行扩容，扩容为原来的2倍，并在扩容之后插入对应的元素。插入一个key/value对后，需要将统计Segment中元素个数的count属性加1。最后，插入成功之后，需要使用unLock()释放锁。

Segment的put函数s.put(key, hash, value, false);

final V put(K key, int hash, V value, boolean onlyIfAbsent) 
{//加锁HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);V oldValue;try {HashEntry<K,V>[] tab = table;int index = (tab.length - 1) & hash;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))) {oldValue = e.value;if (!onlyIfAbsent) {e.value = value;++modCount;}break;}e = e.next;}else {if (node != null)node.setNext(first);elsenode = new HashEntry<K,V>(hash, key, value, first);int c = count + 1;if (c > threshold && tab.length < MAXIMUM_CAPACITY)rehash(node);elsesetEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {unlock();}return oldValue;
}

put操作要加锁，通过计算hash与Segment中HashEntry数组的大小-1的值进行按位与，可以得到元素在对应Segment中的HashEntry中的位置

同样的，put方法首先也会通过hash算法定位到对应的Segment，此时，如果获取到的Segment为空，则调用ensureSegment()方法

size函数

ConcurrentHashMap的size操作的实现方法也非常巧妙，一开始并不对Segment加锁，而是直接尝试将所有的Segment元素中的count相加，这样执行两次，然后将两次的结果对比，如果两次结果相等则直接返回；而如果两次结果不同，则再将所有Segment加锁，然后再执行统计得到对应的size值。

public int size() {// Try a few times to get accurate count. On failure due to// continuous async changes in table, resort to locking.final Segment<K,V>[] segments = this.segments;int size;boolean overflow; // true if size overflows 32 bitslong sum;         // sum of modCountslong last = 0L;   // previous sumint retries = -1; // first iteration isn't retrytry {for (;;) {if (retries++ == RETRIES_BEFORE_LOCK) {for (int j = 0; j < segments.length; ++j)ensureSegment(j).lock(); // force creation}sum = 0L;size = 0;overflow = false;for (int j = 0; j < segments.length; ++j) {Segment<K,V> seg = segmentAt(segments, j);if (seg != null) {sum += seg.modCount;int c = seg.count;if (c < 0 || (size += c) < 0)overflow = true;}}if (sum == last)break;last = sum;}} finally {if (retries > RETRIES_BEFORE_LOCK) {for (int j = 0; j < segments.length; ++j)segmentAt(segments, j).unlock();}}return overflow ? Integer.MAX_VALUE : size;
}

1.8

在JDK1.7之前，ConcurrentHashMap是通过分段锁机制来实现的，所以其最大并发度受Segment的个数限制。因此，在JDK1.8中，ConcurrentHashMap的实现原理摒弃了这种设计，而是选择了与HashMap类似的数组+链表+红黑树的方式实现，而加锁则采用CAS和synchronized实现。

可能写的还不错
可能写的还不错

CAS解析

一般地，锁分为悲观锁和乐观锁：悲观锁认为对于同一个数据的并发操作，一定是为发生修改的；而乐观锁则任务对于同一个数据的并发操作是不会发生修改的，在更新数据时会采用尝试更新不断重试的方式更新数据。

CAS（Compare And Swap，比较交换）：CAS有三个操作数，内存值V、预期值A、要修改的新值B，当且仅当A和V相等时才会将V修改为B，否则什么都不做。Java中CAS操作通过JNI本地方法实现，在JVM中程序会根据当前处理器的类型来决定是否为cmpxchg指令添加lock前缀。如果程序是在多处理器上运行，就为cmpxchg指令加上lock前缀（Lock Cmpxchg）；反之，如果程序是在单处理器上运行，就省略lock前缀。

Intel的手册对lock前缀的说明如下：

• 1、确保对内存的读-改-写操作原子执行。之前采用锁定总线的方式，但开销很大；后来改用缓存锁定来保证指令执行的原子性。
• 2、禁止该指令与之前和之后的读和写指令重排序。
• 3、把写缓冲区中的所有数据刷新到内存中。

CAS同时具有volatile读和volatile写的内存语义。

不过CAS操作也存在一些缺点：

• 1、 存在ABA问题，其解决思路是使用版本号；
• 2、 循环时间长，开销大；
• 3、 只能保证一个共享变量的原子操作。

为了能更好的利用CAS原理解决并发问题，JDK1.5之后在java.util.concurrent.atomic包下采用CAS实现了一系列的原子操作类。

ConcurrentHashMap的数据结构(数组+链表+红黑树)

JDK1.8的ConcurrentHashMap数据结构比JDK1.7之前的要简单的多，其使用的是HashMap一样的数据结构：数组+链表+红黑树。ConcurrentHashMap中包含一个table数组，其类型是一个Node数组；而Node是一个继承自Map.Entry<K, V>的链表，而当这个链表结构中的数据大于8，则将数据结构升级为TreeBin类型的红黑树结构。

另外，JDK1.8中的ConcurrentHashMap中还包含一个重要属性sizeCtl，其是一个控制标识符，不同的值代表不同的意思：

• 其为0时，表示hash表还未初始化，
• 而为正数时这个数值表示初始化或下一次扩容的大小，相当于一个阈值；即如果hash表的实际大小>=sizeCtl，则进行扩容，默认情况下其是当前ConcurrentHashMap容量的0.75倍；
• 而如果sizeCtl为-1，表示正在进行初始化操作；而为-N时，则表示有N-1个线程正在进行扩容。

ConcurrentHashMap的初始化

所有的构造函数都会调用下面的这个构造函数：

public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException();if (initialCapacity < concurrencyLevel)   // Use at least as many binsinitialCapacity = concurrencyLevel;   // as estimated threadslong size = (long)(1.0 + (long)initialCapacity / loadFactor);int cap = (size >= (long)MAXIMUM_CAPACITY) ?MAXIMUM_CAPACITY : tableSizeFor((int)size);this.sizeCtl = cap; //Node数组的初始大小
}

该初始化过程通过指定的初始容量initialCapacity，加载因子loadFactor和预估并发度concurrencyLevel三个参数计算table数组的初始大小sizeCtl的值。

可以看到，在构造ConcurrentHashMap时，并不会对hash表（Node<K, V>[] table）进行初始化，hash表的初始化是在插入第一个元素时进行的。在put操作时，如果检测到table为空或其长度为0时，则会调用initTable()方法对table进行初始化操作。

private final Node<K,V>[] initTable() 
{Node<K,V>[] tab; int sc;while ((tab = table) == null || tab.length == 0) {//sizeCtl在这里赋值给变量scif ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin//SIZECTL = U.objectFieldOffset(k.getDeclaredField("sizeCtl"));else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if ((tab = table) == null || tab.length == 0) {int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = tab = nt;sc = n - (n >>> 2);}} finally {sizeCtl = sc;}break;}}return tab;
}

SIZECTL = U.objectFieldOffset(k.getDeclaredField(“sizeCtl”));

可以看到，该方法使用一个循环实现table的初始化；在循环中，首先会判断sizeCtl的值，如果其小于0，则说明其正在进行初始化或扩容操作，则不执行任何操作，调用yield()方法使当前线程返回等待状态；

而如果sizeCtl大于等于0，则进行cas操作，比较取出SIZECTL位置的元素值，查看是否和sc相等，U.compareAndSwapInt(this, SIZECTL, sc, -1)，如果相等修改成-1，表示正在进行初始化操作。初始化时，如果sizeCtl的值为0，则创建默认容量的table；否则创建大小为sizeCtl的table；然后重置sizeCtl的值为0.75n，即当前table容量的0.75倍，并返回创建的table，此时初始化hash表完成。

Node链表和红黑树结构转换

上文中说到，一个table元素会根据其包含的Node节点数在链表和红黑树两种结构之间切换，因此我们本节先介绍Node节点的结构转换的实现。

首先，在table中添加一个元素时，如果添加元素的链表节点个数超过8，则会触发链表向红黑树结构转换。具体的实现方法如下：

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;elsetl.next = p;tl = p;}setTabAt(tab, index, new TreeBin<K,V>(hd));}}}}
}

该方法首先会检查hash表的大小是否大于等于MIN_TREEIFY_CAPACITY，默认值为64，如果小于该值，则表示不需要转化为红黑树结构，直接将hash表扩容即可。

如果当前table的长度大于64，则使用CAS获取指定的Node节点，然后对该节点通过synchronized加锁，由于只对一个Node节点加锁，因此该操作并不影响其他Node节点的操作，因此极大的提高了ConcurrentHashMap的并发效率。加锁之后，便是将这个Node节点所在的链表转换为TreeBin结构的红黑树。

然后，在table中删除元素时，如果元素所在的红黑树节点个数小于6，则会触发红黑树向链表结构转换。具体实现如下：

static <K,V> Node<K,V> untreeify(Node<K,V> b) {Node<K,V> hd = null, tl = null;for (Node<K,V> q = b; q != null; q = q.next) {Node<K,V> p = new Node<K,V>(q.hash, q.key, q.val, null);if (tl == null)hd = p;elsetl.next = p;tl = p;}return hd;
}

该方法实现简单，在此不再进行细致分析。

ConcurrentHashMap的操作

get操作

通过get获取hash表中的值时，首先需要获取key值的hash值。而在JDK1.8的ConcurrentHashMap中通过speed()方法获取。

static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;
}

speed()方法将key的hash值进行再hash，让hash值的高位也参与hash运算，从而减少哈希冲突。然后再查询对应的value值。

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;int h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}else if (eh < 0)return (p = e.find(h, key)) != null ? p.val : null;while ((e = e.next) != null) {if (e.hash == h &&((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}return null;
}

查询时，首先通过tabAt()方法找到key对应的Node链表或红黑树，然后遍历该结构便可以获取key对应的value值。其中，tabAt()方法主要通过Unsafe类的getObjectVolatile()方法获取value值，通过volatile读获取value值，可以保证value值的可见性，从而保证其是当前最新的值。

put操作

JDK1.8的ConcurrentHashMap的put操作实现方式主要定义在putVal(K key, V value, boolean onlyIfAbsent)中。

final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();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)tab = initTable();else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {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)tab = helpTransfer(tab, f);else {V oldVal = null;synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) {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) {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)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}addCount(1L, binCount);return null;
}

put操作大致可分为以下几个步骤：

• 计算key的hash值，即调用speed()方法计算hash值；
• 获取hash值对应的Node节点位置，此时通过一个循环实现。有以下几种情况：
• 如果table表为空，则首先进行初始化操作，初始化之后再次进入循环获取Node节点的位置；
• 如果table不为空，但没有找到key对应的Node节点，则直接调用casTabAt()方法插入一个新节点，此时不用加锁；
• 如果table不为空，且key对应的Node节点也不为空，但Node头结点的hash值为MOVED(-1)，则表示需要扩容，此时调用helpTransfer()方法进行扩容；
• 其他情况下，则直接向Node中插入一个新Node节点，此时需要对这个Node链表或红黑树通过synchronized加锁。
• 插入元素后，判断对应的Node结构是否需要改变结构，如果需要则调用treeifyBin()方法将Node链表升级为红黑树结构；
• 最后，调用addCount()方法记录table中元素的数量。

size操作

JDK1.8的ConcurrentHashMap中保存元素的个数的记录方法也有不同，首先在添加和删除元素时，会通过CAS操作更新ConcurrentHashMap的baseCount属性值来统计元素个数。但是CAS操作可能会失败，因此，ConcurrentHashMap又定义了一个CounterCell数组来记录CAS操作失败时的元素个数。因此，ConcurrentHashMap中元素的个数则通过如下方式获得：

元素总数 = baseCount + sum(CounterCell)

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)sum += a.value;}}return sum;
}

而JDK1.8中提供了两种方法获取ConcurrentHashMap中的元素个数。

public int size() {long n = sumCount();return ((n < 0L) ? 0 :(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :(int)n);
}public long mappingCount() {long n = sumCount();return (n < 0L) ? 0L : n; // ignore transient negative values
}

如代码所示，size只能获取int范围内的ConcurrentHashMap元素个数；而如果hash表中的数据过多，超过了int类型的最大值，则推荐使用mappingCount()方法获取其元素个数。

以上主要分析了ConcurrentHashMap在JDK1.7和JDK1.8中的两种不同实现方案，当然ConcurrentHashMap的功能强大，还有很多方法本文都未能详细解析，但其分析方法与本文以上的内容类似，因此不再赘述，感兴趣的同学可以自行分析比较。通过学习JDK源码，对以后的Java程序设计也有一定的帮助。本系列文章将深入剖析Java concurrent包中的并发编程设计，并从中提炼出一些使用场景，从而为今后的Java程序设计提供一些小小的灵感。