本篇文章将从concurrentHashMap的源码出发，带大家一点点了解它的实现原理。

文章结构

• 部分关键参数
• 构造方法
• put
• get
• 总结

首先简单说说他的特性。

1. concurrentHashMap跟Hashtable具有相同的功能方法。可以看作是Hashtable的升级版，HashMap的线程安全版。

2. 跟Hashtable相同，它的键或值不允许是null。

3. ConcurrentHashMap和HashMap一样都是采用拉链法处理哈希冲突，且都为了防止单链表过长影响查询效率，所以当链表长度超过某一个值时候将用红黑树代替链表进行存储，采用了数组+链表+红黑树的结构

4. ConcurrentHashMap与Hashtable比较

• 线程安全的实现：Hashtable采用对象锁(synchronized修饰对象方法)来保证线程安全，也就是一个Hashtable对象只有一把锁，如果线程1拿了对象A的锁进行有synchronized修饰的put方法，其他线程是无法操作对象A中有synchronized修饰的方法的(如get方法、remove方法等)，竞争激烈所以效率低下。而ConcurrentHashMap采用CAS + synchronized来保证并发安全性，且synchronized关键字不是用在方法上而是用在了具体的对象上，实现了更小粒度的锁，等会源码分析的时候在细说这个SUN大师们的鬼斧神工
• 数据结构的实现：Hashtable采用的是数组 + 链表，当链表过长会影响查询效率，而ConcurrentHashMap采用数组 + 链表 + 红黑树，当链表长度超过某一个值，则将链表转成红黑树，提高查询效率。

ConcurrentHashMap

我们先看一下ConcurrentHashMap实现了哪些接口、继承了哪些类，对ConcurrentHashMap有一个整体认知。

ConcurrentHashMap跟HashMap一样继承AbstractMap接口，

然后实现了ConcurrentMap接口，这个和HashMap不一样，HashMap是直接实现的Map接口。

部分关键参数

//node数组最大容量：2^30=1073741824  
private static final int MAXIMUM_CAPACITY = 1 << 30;//默认初始表容量。必须是 2 的幂（即至少为 1）且最多为 MAXIMUM_CAPACITY
private static final int DEFAULT_CAPACITY = 16;//最大可能（非二的幂）数组大小。 toArray 和相关方法需要
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;//负载因子
private static final float LOAD_FACTOR = 0.75f;// 链表转红黑树阀值,> 8 链表转换为红黑树  
static final int TREEIFY_THRESHOLD = 8;//树转链表阀值，小于等于6（tranfer时，lc、hc=0两个计数器分别++记录原bin、新binTreeNode数量，<=UNTREEIFY_THRESHOLD 则untreeify(lo)）
static final int UNTREEIFY_THRESHOLD = 6;//存放node的数组  transient  volatile  Node<K,V>[] table;  /*控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义  *当为负数时：-1 代表正在初始化，-N 代表有N-1 个线程正在 进行扩容  *当为  0  时：代表当时的table还没有被初始化  *当为正数时：表示初始化或者下一次进行扩容的大小  
*/
private  transient  volatile  int  sizeCtl;  

几个重要的成员变量table、nextTable、baseCount、sizeCtl、transferIndex、cellsBusy

• table：数据类型是Node数组，这里的Node和HashMap的Node一样都是内部类且实现了Map.Entry接口
• nextTable：哈希表扩容时生成的数据，数组为扩容前的2倍
• sizeCtl：多个线程的共享变量，是操作的控制标识符，它的作用不仅包括threshold的作用，在不同的地方有不同的值也有不同的用途。
  • -1代表正在初始化
  • -N代表有N-1个线程正在进行扩容操作
  • 0代表hash表还没有被初始化
  • 正数表示下一次进行扩容的容量大小
• ForwardingNode：一个特殊的Node节点，Hash地址为-1，存储着nextTable的引用，只有table发生扩用的时候，ForwardingNode才会发挥作用，作为一个占位符放在table中表示当前节点为null或者已被移动

ConcurrentHashMap 针对 ForwardingNode、ReservationNode，以及树根结点都定义了特定的哈希值：

下面判断用到的很多哦。

/** 节点哈希字段的编码. See above for explanation.*/
/** ForwardingNode 结点的 hash 值 */
static final int MOVED     = -1; // hash for forwarding nodes  转发节点的哈希static final int TREEBIN   = -2; // hash for roots of trees    树根的哈希static final int RESERVED  = -3; // hash for transient reservations  临时预订的哈希
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

构造方法

它的构造方法一共有5个，从数量上看就和HashMap、Hashtable(4个)的不同，多出的那个构造函数是public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)，即除了传入容量大小、负载因子之外还多传入了一个整型的concurrencyLevel，这个整型是我们预先估计的并发量，比如我们估计并发是30，那么就可以传入30。

其他的4个构造函数的参数和HashMap的一样，而具体的初始化过程却又不相同，HashMap和Hashtable传入的容量大小和负载因子都是为了计算出初始阈值(threshold)，而ConcurrentHashMap传入的容量大小和负载因子是为了计算出sizeCtl用于初始化table，这个sizeCtl即table数组的大小，不同的构造函数计算sizeCtl方法都不一样。

//1.  无参构造函数，什么也不做，此时sizeCtl参数为
//table的初始化放在了第一次插入数据时，默认容量大小是16。
//使用默认初始表大小 (16) 创建一个新的空映射
public ConcurrentHashMap() {
}
//2.  传入容量大小的构造方法
public ConcurrentHashMap(int initialCapacity) {if (initialCapacity < 0)throw new IllegalArgumentException();//如果传入的容量大小大于允许的最大容量值 则cap取允许的容量最大值 否则cap =//((传入的容量大小 + 传入的容量大小无符号右移1位 + 1)的结果向上取最近的2幂次方)，//即如果传入的容量大小是12 则 cap = 32 //(12 >>> 1)=1100>>>1=0110=7//=(12 + (12 >>> 1) + 1=19 并向上取2的幂次方即32)//，这里为啥一定要是2的幂次方，原因和HashMap的threshold一样，//都是为了让位运算和取模运算的结果一样。//MAXIMUM_CAPACITY即允许的最大容量值 为2^30。int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?MAXIMUM_CAPACITY :tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));//将上面计算出的cap 赋值给sizeCtl，注意此时sizeCtl为正数，代表进行扩容的容量大小。this.sizeCtl = cap;
}//3.创建一个与给定Map具有相同映射Mappublic ConcurrentHashMap(Map<? extends K, ? extends V> m) {//置sizeCtl为默认容量大小 即16。this.sizeCtl = DEFAULT_CAPACITY;putAll(m);
}//4.传入容量大小和负载因子的构造方法
//默认并发数大小是1。
public ConcurrentHashMap(int initialCapacity, float loadFactor) {this(initialCapacity, loadFactor, 1);
}//5. 传入容量大小、负载因子和并发数大小的构造方法	
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {//如果负载因子>0，传入容量大小<0,并发数大小<=0一项不符合就//直接抛出IllegalArgumentException非法参数异常if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException();//如果传入的容量大小小于并发数大小，咱们就把初始容量变成更大的并发数。这样做的原因是确保每一个Node只会分配给一个线程if (initialCapacity < concurrencyLevel)   // initialCapacity = concurrencyLevel;   // 
/**
下面就时计算sizeCtl的值了，用于下一次扩容的时候就是初始化的时候。
*/long size = (long)(1.0 + (long)initialCapacity / loadFactor);//如果size比允许的最大容量值还大，那直接sizeCtl= 允许的最大容量值。// 否则对size进行tableSizeFor操作就是 size向上取2的幂次方//比如 size = 6   tableSizeFor((int)size)=9//    size = 9   tableSizeFor((int)size)=16//    size = 12   tableSizeFor((int)size)=32int cap = (size >= (long)MAXIMUM_CAPACITY) ?MAXIMUM_CAPACITY : tableSizeFor((int)size)=9;this.sizeCtl = cap;  //计算好之后便赋值给sizeCtl
}

构造方法简单总结：

都是为了算出sizeCtl值。就是初始化table数组大小。

一共有5个构造方法。

1. 无参构造函数，什么也不做，此时sizeCtl参数为0，初始化操作在put操作中实现

2. 传入容量大小的构造方法，对参数校验后(>0)，传入的容量大小大于MAXIMUM_CAPACITY即允许的最大容量 2^30， 则cap取允许的容量最大值。比他小咱们就做tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1))操作，对initialCapacity + (initialCapacity >>> 1) + 1 向上取最近的2的幂次方，最后将值给sizeCtl。

3. 创建一个与给定Map具有相同映射Map。

4. 传入容量大小和负载因子的构造方法，它调用第5个方法，并发数大小=1。

5. 传入容量大小、负载因子和预期并发数大小的构造方法。具体看上面。

put操作

put

public V put(K key, V value) {return putVal(key, value, false);//调用了putVal方法
}

它就调用了putVal方法，咱们来看putVal方法

putVal

final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();//如果有一个为null，那么直接报错。所以可以得出concurrentHashMap中Key，value不能为空。int hash = spread(key.hashCode());//计算key的hash值，参与插入位置的计算int binCount = 0;for (Node<K,V>[] tab = table;;) {//循环尝试,for循环内它是if..。else if...else格式Node<K,V> f; int n, i, fh;//如果table数组为null，说明还没有table数组呢，那咱们就开始初始化数组。if (tab == null || (n = tab.length) == 0)//初始化之后，会直接进行下次循环。tab = initTable();//如果i位置没有数据(即该下标上的链表为空)，就直接CAS无锁插入else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//获取对应位置Node节点(链表头节点)(Volatile的)// casTabAt方法CAS无锁的添加到空箱,  如果插入成功了则跳出for循环，插入//失败(其他线程抢先插入了)，那么继续下次循环。if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break;               }//如果该下标上的节点的哈希地址为-1(即链表的头节点为ForwardingNode节点)，则表示//table正在进行扩容（transfer），那咱们先等着吧。//注意的：：ConcurrentHashMap初始化和扩容不是用同一个方法，而//HashMap和Hashtable都是用同一个方法，并且当前线程会去协助扩容，扩容过程后面介绍。else if ((fh = f.hash) == MOVED)//-1tab = helpTransfer(tab, f);//返回扩容完成后的table。//上面情况都没有，那么我们就将进入到链表中，将新节点插入或者覆盖旧值。else {V oldVal = null;// only针对首个节点进行加锁操作，一次锁的资源小了，减少了锁的粒度。也正是因为这个提高了ConcurrentHashMap的效率，提高了并发度。synchronized (f) {if (tabAt(tab, i) == f) {  //做一次校验看看头节点一样不if (fh >= 0) {//哦~不知道fh怎么来的？看26行。节点的哈希地址大于等于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)))) {//如果发现有key一样的，那咱们就把value值替换掉。oldVal = e.val;if (!onlyIfAbsent)e.val = value;break;//结束推出链表循环}// 如果没有找到值为key的节点，直接新建Node并加入链表尾部即可。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) {//如果首节点为TreeBin类型，说明为红黑树结构，执行putTreeVal操作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)//如果链表的长度大于树形阈值 8 时就会进行红黑树的转换treeifyBin(tab, i); if (oldVal != null)return oldVal;//如果key对应的以前有值，那么返回oldValbreak;}}}addCount(1L, binCount);//统计size，可能会扩容transfer操作。return null;
}

好！咱们先简单总结一下put流程：

1. 如果没有初始化就先调用initTable（）方法来进行初始化过程

2. 如果没有hash冲突就直接CAS插入

3. 如果还在进行扩容操作transfe-ing就先进行扩容,等它扩容完了咱们再尝试插入。

4. 如果存在hash冲突，就对首节点加synchronized锁来保证线程安全，这里有两种情况，一种是链表形式就直接遍历到尾端插入，一种是红黑树就按照红黑树结构插入，结束加锁。

5. 最后一个如果Hash冲突时会形成Node链表，在链表长度超过8，Node数组超过64时会将链表结构转换为红黑树的结构，break退出循环

6. 如果添加成功就调用addCount（）方法统计ConcurrentHashMap的size，并加一，检查是否需要扩容，可能会扩容transfer操作。

了解了它的基本流程之后，我们来看其中的一些调用方法。

initTable     初始化数组
tabAt		  获取对应位置Node节点
casTabAt	  CAS添加节点helpTransfer  	帮助扩容，如果线程进入到这边说明已经有其他线程正在做扩容操作，这个是一个辅助方法
transfer   		扩容操作
addCount  		统计ConcurrentHashMap的size，并加一

咱们一个个说。

initTable

当table数组是null的时候调用。

initTable()方法初始化一个合适大小的数组，然后设置sizeCtl。

我们知道ConcurrentHashMap是线程安全的，即支持多线程的，那么一开始很多个线程同时执行put()方法，而table又没初始化，那么就会很多个线程会去执行initTable()方法尝试初始化table，而put方法和initTable方法都是没有加锁的(synchronize)，

那SUN的大师们是怎么保证线程安全的呢？

/*** 使用 sizeCtl 中记录的大小初始化表  table。*/
private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;//先看看table是否被初始化，如果已经初始化了，结束while循环//如果table为null那么一直while循环尝试初始化，直到完成while ((tab = table) == null || tab.length == 0) {\if ((sc = sizeCtl) < 0) // sizeCtl< 0说明，其他线程抢先的正在初始化或者扩容，Thread.yield(); 	// 初始化竞争失败；让出cpu，等着就完了。运行态转成就绪态//sizeCtl >= 0else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //Unsafed.compareAndSwapInt以CAS操作CASsizeCtl=-1,表示初始化状态。try {if ((tab = table) == null || tab.length == 0) {//再次确认当前table为null即还未初始化，这个判断不能少。//如果sc(sizeCtl)大于0，则n=sc，否则n=默认的容量大小16，//这里的sc=sizeCtl=0，即如果在构造函数没有指定容量大小，//否则使用了有参数的构造函数，sc=sizeCtl=指定的容量大小。int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; //new Node数组(table)table = tab = nt;sc = n - (n >>> 2); //计算阈值，n - (n >>> 2) = 0.75*n，大于这个阈值，就会发生扩容。}} finally {sizeCtl = sc;}break;}}return tab;
}

tabAt AND casTabAt

都是调用了Unsafe类中的方法。大家感兴趣可以看我的这篇文章

@SuppressWarnings("unchecked")
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);
}static 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);
}

helpTransfer 和 transfer

/*** Helps transfer if a resize is in progress.*/
//协助扩容方法
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {Node<K,V>[] nextTab; int sc;//如果当前table不为null 且 f为ForwardingNode节点 且新的table即nextTable存在的情况下才能协助扩容，该方法的作用是让线程参与扩容的复制。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) {	//如果小于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)) { //更新sizeCtl的值，+1，代表新增一个线程参与扩容transfer(tab, nextTab);break;}}return nextTab;}return table;
}

transfer()方法为ConcurrentHashMap扩容操作的核心方法。由于ConcurrentHashMap支持多线程扩容，而且也没有进行加锁，所以实现会变得有点儿复杂。整个扩容操作分为两步：

1. 构建一个nextTable，其大小为原来大小的两倍，这个步骤是在单线程环境下完成的
2. 将原来table里面的内容复制到nextTable中，这个步骤是允许多线程操作的，所以性能得到提升，减少了扩容的时间消耗。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {int n = tab.length, stride;//根据服务器CPU数量来决定每个线程负责的bucket桶数量，避免因为扩容的线程过多反而影响性能。//如果CPU数量为1，则stride=1，否则将需要迁移的bucket数量(table大小)除以CPU数量，平分给//各个线程，但是如果每个线程负责的bucket数量小于限制的最小是(16)的话，则强制给每个线程//分配16个bucket数。if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide range//如果nextTable还未初始化，则初始化nextTable，这个初始化和iniTable初始化一样，只能由//一个线程完成。if (nextTab == null) {   // nextTab为null，那么就初始化它try {@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) {      // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;transferIndex = n;}int nextn = nextTab.length;//构建一个连节点的指针，用于标识位ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);boolean advance = true;//循环的关键变量，判断是否已经扩容完成，完成就return，退出循环boolean finishing = false; // to ensure sweep before committing nextTab//下个循环是分配任务和控制当前线程的任务进度，这部分是transfer()的核心逻辑，描述了如何与其他线//程协同工作。for (int i = 0, bound = 0;;) { //进入循环Node<K,V> f; int fh;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;}}//i<0说明已经遍历完旧的数组tab；i>=n什么时候有可能呢？在下面看到i=n,所以目前i最大应该是n吧。//i+n>=nextn,nextn=nextTab.length，所以如果满足i+n>=nextn说明已经扩容完成if (i < 0 || i >= n || i + n >= nextn) {int sc;if (finishing) {nextTable = null;table = nextTab; sizeCtl = (n << 1) - (n >>> 1);return;}//利用CAS方法更新这个扩容阈值，在这里面sizectl值减一，说明新加入一个线程参与到扩容操作,参考sizeCtl的注释if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;finishing = advance = true;//finishing和advance保证线程已经扩容完成了可以退出循环i = n; //先退出，重新检查遍}}else if ((f = tabAt(tab, i)) == null)//如果tab[i]为null，那么就把fwd插入到tab[i]，表明这个节点已经处理过了advance = casTabAt(tab, i, null, fwd);else if ((fh = f.hash) == MOVED)//那么如果f.hash=-1的话说明该节点为ForwardingNode，说明该节点已经处理过了advance = true; // already processed//迁移过程（对当前指向的bucket），这部分的逻辑与HashMap类似，拿旧数组的容量当做一//个掩码，然后与节点的hash进行与&操作，可以得出该节点的新增有效位，如果新增有效位为//0就放入一个链表A，如果为1就放入另一个链表B，链表A在新数组中的位置不变（跟在旧数//组的索引一致），链表B在新数组中的位置为原索引加上旧数组容量。else {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);elsehn = new Node<K,V>(ph, pk, pv, hn);}//setTabAt(nextTab, i, ln);setTabAt(nextTab, i + n, hn);setTabAt(tab, i, fwd);//把已经替换的节点的旧tab的i的位置用fwd结点替换，fwd包含nextTabadvance = 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;elseloTail.next = p;loTail = p;++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.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;}}}}}
}

扩容总结

首先根据运算得到需要遍历的次数i，然后利用tabAt方法获得i位置的元素：

1. 如果这个位置为空，就在原table中的i位置放入forwardNode节点，这个也是触发并发扩容的关键点；
2. 如果这个位置是Node节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把他们分别放在nextTable的i和i+n的位置上
3. 如果这个位置是TreeBin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi，把处理的结果分别放在nextTable的i和i+n的位置上
4. 遍历过所有的节点以后就完成了复制工作，这时让nextTable作为新的table，并且更新sizeCtl为新容量的0.75倍，完成扩容。

多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。

get操作

get

返回指定键映射到的值，如果此映射不包含该键的映射，则返回null 。

和HashMap一样用equals来判断是否相等，也就是说必须Hash索引和equals一样，才算相同。

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;int h = spread(key.hashCode());//运用key的hashCode()计算出哈希地址//table不为空 且 table长度大于0 且 计算出的下标上用tabAt CAS取值不为空，也就是这个位置上有人，那么就进去找。if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {  //如果哈希地址、键key相同则表示查找到，返回value，这里查找到的是头节点，就不用往下找了。if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}//eh=-2ash值<0 可能是Tree树形，可能e节点为ForwardingNode结点。//如果eh=-1就说明e节点为ForwardingNode,这说明什么，说明这个节点已经不存在了，被另一个线程正则扩容所以要查找key对应的值的话，直接到新newtable找//如果是eh=-2 说明e节点为Tree根结点，那么后面就以红黑树的方式查找。else if (eh < 0)return (p = e.find(h, key)) != null ? p.val : null;//eh=-1这里调用的find其实就是ForwardingNode中的find方法了。//排除完上面情况，咱就正常找while ((e = e.next) != null) {if (e.hash == h &&((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}return null;
}

总结流程：

1. 调用spread()方法计算key的hashCode()获得哈希地址。
2. 计算出键key所在的下标，算法是(n - 1) & h，如果table不为空，且下标上的bucket不为空，则到bucket中查找。
3. 如果bucket的头节点的哈希地址小于0，说明e节点可能为ForwardingNode,这说明什么，说明这个节点已经不存在了，被另一个线程正则扩容所以要查找key对应的值的话，直接到新newtable找，如果e结点是树的根节点，那么就按照红黑树的方式查找。
4. 找到则返回该键key的值，找不到则返回null。

可能有人要问了？put的时候加锁了，get读的时候不需要锁吗？

这要归功于使用的tabAt中的了Unsafe的getObjectVolatile，因为table是volatile类型，所以对tab[i]的原子请求也是可见的。因为如果同步正确的情况下，根据happens-before原则，对volatile域的写入操作happens-before于每一个后续对同一域的读操作。所以不管其他线程对table链表或树的修改，都对get读取可见。

总结与思考

其实可以看出JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap，相对而言，ConcurrentHashMap只是增加了同步的操作来控制并发，从JDK1.7版本的ReentrantLock+Segment+HashEntry，到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言，总结如下思考

其实可以看出JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap，相对而言，ConcurrentHashMap只是增加了同步的操作来控制并发，从JDK1.7版本的ReentrantLock+Segment+HashEntry，到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言，总结如下思考

1. JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）
2. JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了
3. JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档
4. JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock，我觉得有以下几点
   1. 因为粒度降低了，在相对而言的低粒度加锁方式，synchronized并不比ReentrantLock差，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势就没有了
   2. JVM的开发团队从来都没有放弃synchronized，而且基于JVM的synchronized优化空间更大，使用内嵌的关键字比使用API更加自然
   3. 在大量的数据操作下，对于JVM的内存压力，基于API的ReentrantLock会开销更多的内存，虽然不是瓶颈，但是也是一个选择依据

打完收工！！！

下面是我收集的一些面试题：

JDK1.8中的ConcurrentHashMap是如何保证线程安全的？

模板2：

1. 储存Map数据的数组时被volatile关键字修饰，一旦被修改，其他线程就可见修改。因为是数组存储，所以只有改变数组内存值是才会触发volatile的可见性
2. 如果put操作时hash计算出的槽点内没有值，采用自旋+CAS保证put一定成功，且不会覆盖其他线程put的值
3. 如果put操作时节点正在扩容，即发现槽点为转移节点，会等待扩容完成后再进行put操作，保证扩容时老数组不会变化
4. 对槽点进行操作时会锁住槽点，保证只有当前线程能对槽点上的链表或红黑树进行操作
5. 红黑树旋转时会锁住根节点，保证旋转时线程安全

2.JDK7和JDK8中的ConcurrentHashMap不同点。

3.扩容期间在未迁移到的hash桶插入数据会发生什么？

答：只要插入的位置扩容线程还未迁移到，就可以插入，当迁移到该插入的位置时，就会阻塞等待插入操作完成再继续迁移 。

4.1 正在迁移的hash桶遇到 get 操作会发生什么？

答：在扩容过程期间形成的 hn 和 ln链 是使用的类似于复制引用的方式，也就是说 ln 和 hn 链是复制出来的，而非原来的链表迁移过去的，所以原来 hash 桶上的链表并没有受到影响，因此如果当前节点有数据，还没迁移完成，此时不影响读，能够正常进行。

如果当前链表已经迁移完成，那么头节点会被设置成fwd节点，此时get线程会帮助扩容。

4.2 正在迁移的hash桶遇到 put/remove 操作会发生什么？

如果当前链表已经迁移完成，那么头节点会被设置成fwd节点，此时写线程会帮助扩容，如果扩容没有完成，当前链表的头节点会被锁住，所以写线程会被阻塞，直到扩容完成。

5.如果 lastRun 节点正好在一条全部都为高位或者全部都为低位的链表上，会不会形成死循环？

答：在数组长度为64之前会导致一直扩容，但是到了64或者以上后就会转换为红黑树，因此不会一直死循环 。

6.扩容后 ln 和 hn 链不用经过 hash 取模运算，分别被直接放置在新数组的 i 和 n + i 的位置上，那么如何保证这种方式依旧可以用过 h & (n - 1) 正确算出 hash 桶的位置？

答：如果 fh & n-1 = i ，那么扩容之后的 hash 计算方法应该是 fh & 2n-1 。 因为 n 是 2 的幂次方数，所以 如果 n=16， n-1 就是 1111(二进制)， 那么 2n-1 就是 11111 (二进制) 。 其实 fh & 2n-1 和 fh & n-1 的值区别就在于多出来的那个 1 => fh & (10000) 这个就是两个 hash 的区别所在 。而 10000 就是 n 。所以说 如果 fh 的第五 bit 不是 1 的话 fh & n = 0 => fh & 2n-1 == fh & n-1 = i 。 如果第5位是 1 的话 。fh & n = n => fh & 2n-1 = i+n 。

7.并发情况下，各线程中的数据可能不是最新的，那为什么 get 方法不需要加锁？

答：get操作全程不需要加锁是因为Node的成员val是用volatile修饰的，在多线程环境下线程A修改结点的val或者新增节点的时候是对线程B可见的。。

8.1 ConcurrentHashMap 和 Hashtable 的区别?

ConcurrentHashMap 和 Hashtable 的区别主要体现在实现线程安全的方式上不同。
底层数据结构：
JDK1.7的 ConcurrentHashMap 底层采用 分段的数组+链表 实现，JDK1.8 采用的数据结构跟HashMap1.8的结构一样，数组+链表/红黑二叉树。Hashtable是采用 数组+链表 的形式。
实现线程安全的方式（重要）：

① 在JDK1.7的时候，ConcurrentHashMap（分段锁） 对整个桶数组进行了分割分段(Segment)，每一把锁只锁容器其中一部分数据，多线程访问容器里不同数据段的数据，就不会存在锁竞争，提高并发访问率。 到了 JDK1.8 的时候已经摒弃了Segment的概念，而是直接用 Node 数组+链表+红黑树的数据结构来实现，并发控制使用 synchronized 和 CAS 来操作。

② Hashtable(同一把锁) :使用 synchronized 来保证线程安全，效率非常低下。当一个线程访问同步方法时，其他线程也访问同步方法，可能会进入阻塞或轮询状态，如使用 put 添加元素，另一个线程不能使用 put 添加元素，也不能使用 get，竞争会越来越激烈效率越低。

8.2 ConcurrentHashMap 和 HashMap 的相同点和不同点

相同之处：

都是数组 +链表+红黑树的数据结构（JDK8之后），所以基本操作的思想一致
都实现了Map接口，继承了AbstractMap 操作类，所以方法大都相似，可以相互切换
不同之处：
ConcurrentHashMap 是线程安全的，多线程环境下，无需加锁直接使用
ConcurrentHashMap 多了转移节点，主要用户保证扩容时的线程安全

9.扩容过程中，读访问能否访问的到数据？怎么实现的？

可以的。当数组在扩容的时候，会对当前操作节点进行判断，如果当前节点还没有被设置成fwd节点，那就可以进行读写操作，如果该节点已经被处理了，那么当前线程也会加入到扩容的操作中去。

10.为什么超过冲突超过8才将链表转为红黑树而不直接用红黑树?

默认使用链表， 链表占用的内存更小
正常情况下，想要达到冲突为8的几率非常小，如果真的发生了转为红黑树可以保证极端情况下的效率

11. ConcurrentHashMap 和HashMap的扩容有什么不同？

HashMap的扩容是创建一个新数组，将值直接放入新数组中，JDK7采用头链接法，会出现死循环，JDK8采用尾链接法，不会造成死循环
ConcurrentHashMap 扩容是从数组队尾开始拷贝，拷贝槽点时会锁住槽点，拷贝完成后将槽点设置为转移节点。所以槽点拷贝完成后将新数组赋值给容器

12. ConcurrentHashMap 是如何发现当前槽点正在扩容的？

ConcurrentHashMap 新增了一个节点类型，叫做转移节点，当我们发现当前槽点是转移节点时（转移节点的 hash 值是 -1），即表示 Map 正在进行扩容.

13.描述一下 CAS 算法在 ConcurrentHashMap 中的应用

• CAS是一种乐观锁，在执行操作时会判断内存中的值是否和准备修改前获取的值相同，如果相同，把新值赋值给对象，否则赋值失败，整个过程都是原子性操作，无线程安全问题
• ConcurrentHashMap 的put操作是结合自旋用到了CAS，如果hash计算出的位置的槽点值为空，就采用CAS+自旋进行赋值，如果赋值是检查值为空，就赋值，如果不为空说明有其他线程先赋值了，放弃本次操作，进入下一轮循环

1、ConcurrentHashMap有哪些构造函数？

一共有五个，

2、ConcurrentHashMap使用什么技术来保证线程安全？

jdk1.7：Segment+HashEntry来进行实现的；

jdk1.8：放弃了Segment臃肿的设计，采用Node+CAS+Synchronized来保证线程安全；

3、ConcurrentHashMap的get方法是否要加锁，为什么？

不需要，get方法采用了unsafe方法，来保证线程安全。

4、ConcurrentHashMap迭代器是强一致性还是弱一致性？HashMap呢？

弱一致性，hashmap强一直性。

ConcurrentHashMap可以支持在迭代过程中，向map添加新元素，而HashMap则抛出了ConcurrentModificationException，

因为HashMap包含一个修改计数器，当你调用他的next()方法来获取下一个元素时，迭代器将会用到这个计数器。

5、ConcurrentHashMap1.7和1.8的区别：

jdk1.8的实现降低锁的粒度，jdk1.7锁的粒度是基于Segment的，包含多个HashEntry，而jdk1.8锁的粒度就是Node

数据结构：jdk1.7 Segment+HashEntry；jdk1.8 数组+链表+红黑树+CAS+synchronized