前言：

强推：一文读懂HashMap

这感觉讲的HashMap很明白。

1. 多线程环境下面，HashMap和Hashtable会怎么样？

1.1 HashMap

因为put会调用：

// 新增Entry。将“key-value”插入指定位置，bucketIndex是位置索引。      void addEntry(int hash, K key, V value, int bucketIndex) {      // 保存“bucketIndex”位置的值到“e”中      Entry<K,V> e = table[bucketIndex];      // 设置“bucketIndex”位置的元素为“新Entry”，      // 设置“e”为“新Entry的下一个节点”      table[bucketIndex] = new Entry<K,V>(hash, key, value, e);      // 若HashMap的实际大小 不小于 “阈值”，则调整HashMap的大小      if (size++ >= threshold)      resize(2 * table.length);      }

现在假如A线程和B线程同时对同一个数组位置调用addEntry，两个线程会同时得到现在的头结点，然后A写入新的头结点之后，B也写入新的头结点，那B的写入操作就会覆盖A的写入操作造成A的写入操作丢失

而且多线程的情况下，HashMap的put操作有可能会出现死锁。(只对于JDK1.7，在JDK1.8中已经修改了，采用后插法，避免了这个问题。)

JDK1.7中使用单链表进行的纵向延伸，采用头插法可以增加插入的效率，但是有可能会形成逆序且形成环。

JDK1.8因为加入了红黑树所以改为了尾插法，避免了链表逆序和死循环。

当多个线程同时操作同一个数组位置的时候，也都会先取得现在状态下该位置存储的头结点，然后各自去进行计算操作，之后再把结果写会到该数组位置去，其实写回的时候可能其他的线程已经就把这个位置给修改过了，就会覆盖其他线程的修改。

    final Entry<K,V> removeEntryForKey(Object key) {      // 获取哈希值。若key为null，则哈希值为0；否则调用hash()进行计算      int hash = (key == null) ? 0 : hash(key.hashCode());      int i = indexFor(hash, table.length);      Entry<K,V> prev = table[i];      Entry<K,V> e = prev;      // 删除链表中“键为key”的元素      // 本质是“删除单向链表中的节点”      while (e != null) {      Entry<K,V> next = e.next;      Object k;      if (e.hash == hash &&      ((k = e.key) == key || (key != null && key.equals(k)))) {      modCount++;      size--;      if (prev == e)      table[i] = next;      else     prev.next = next;      e.recordRemoval(this);      return e;      }      prev = e;      e = next;      }      return e;      }

这个操作会新生成一个新的容量的数组，然后对原数组的所有键值对重新进行计算和写入新的数组，之后指向新生成的数组。

当多个线程同时检测到总数量超过门限值的时候就会同时调用resize操作，各自生成新的数组并rehash后赋给该map底层的数组table，结果最终只有最后一个线程生成的新数组被赋给table变量，其他线程的均会丢失。而且当某些线程已经完成赋值而其他线程刚开始的时候，就会用已经被赋值的table作为原始数组，这样也会有问题。

// 重新调整HashMap的大小，newCapacity是调整后的容量      void resize(int newCapacity) {      Entry[] oldTable = table;      int oldCapacity = oldTable.length;     //如果就容量已经达到了最大值，则不能再扩容，直接返回    if (oldCapacity == MAXIMUM_CAPACITY) {      threshold = Integer.MAX_VALUE;      return;      }      // 新建一个HashMap，将“旧HashMap”的全部元素添加到“新HashMap”中，      // 然后，将“新HashMap”赋值给“旧HashMap”。      Entry[] newTable = new Entry[newCapacity];      transfer(newTable);      table = newTable;      threshold = (int)(newCapacity * loadFactor);      }

转自：HashMap为什么是线程不安全的

1.2 Hashtable

使用synchronized来保证线程安全，但在线程竞争激烈的情况下Hashtable的效率非常低下。因为当一个线程访问Hashtable的同步方法时，其他线程访问Hashtable的同步方法时，可能会进入阻塞或轮询状态。如线程1使用put进行添加元素，线程2不但不能使用put方法添加元素，并且也不能使用get方法来获取元素，所以竞争越激烈效率越低。

2. ConcurrentHashMap

主要了解怎么解决线程安全又可以提高性能呢？

因为Hashtable竞争同一把锁，假如容器里面有多把锁，那么每一把锁锁住一部分数据，那么访问不同数据段的时候就没有锁竞争了。

所以是分段锁来解决的。

有些方法需要跨段，比如size()和containsValue()，它们可能需要锁定整个表而而不仅仅是某个段，需要按照顺序锁住所有的段

操作完毕后，又按顺序释放所有段的锁。

因为不按照顺序容易出现死锁。

在ConcurrentHashMap内部，段数组是final的，并且其成员变量实际上也是final的，但是，仅仅是将数组声明为final的并不保证数组成员也是final的，这需要实现上的保证。这可以确保不会出现死锁，因为获得锁的顺序是固定的。

ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap类似，是一种数组和链表结构，一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素，每个Segment守护者一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时，必须首先获得它对应的Segment锁。

但是！！！！

DK1.8的实现已经抛弃了Segment分段锁机制，利用CAS+Synchronized来保证并发更新的安全。数据结构采用：数组+链表+红黑树。

构造：

    //构造方法public ConcurrentHashMap(int initialCapacity) {if (initialCapacity < 0)//判断参数是否合法throw new IllegalArgumentException();int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?MAXIMUM_CAPACITY ://最大为2^30tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));//根据参数调整table的大小this.sizeCtl = cap;//获取容量//ConcurrentHashMap在构造函数中只会初始化sizeCtl值，并不会直接初始化table}//调整table的大小private static final int tableSizeFor(int c) {//返回一个大于输入参数且最小的为2的n次幂的数。int n = c - 1;n |= n >>> 1;n |= n >>> 2;n |= n >>> 4;n |= n >>> 8;n |= n >>> 16;return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;}

tableSizeFor(int c)的原理：将c最高位以下通过|=运算全部变成1，最后返回的时候，返回n+1；
eg:当输入为25的时候，n等于24，转成二进制为1100，右移1位为0110，将1100与0110进行或("|")操作，得到1110。接下来右移两位得11，再进行或操作得1111，接下来操作n的值就不会变化了。最后返回的时候，返回n+1，也就是10000，十进制为32。按照这种逻辑得到2的n次幂的数。
那么为什么要先-1再+1呢？输入若是为0，那么不论怎么操作，n还是0，但是HashMap的容量只有大于0时才有意义。

table初始化：

table初始化操作会延缓到第一次put行为。但是put是可以并发执行的，那么是如何实现table只初始化一次的？接着上源码：

    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; K fk; V fv;if (tab == null || (n = tab.length) == 0)//判断table还未初始化tab = initTable();//初始化tableelse if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))break;                   // no lock when adding to empty bin}...省略一部分源码}} private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;while ((tab = table) == null || tab.length == 0) {//如果一个线程发现sizeCtl<0，意味着另外的线程执行CAS操作成功，当前线程只需要让出cpu时间片，//由于sizeCtl是volatile的，保证了顺序性和可见性if ((sc = sizeCtl) < 0)//sc保存了sizeCtl的值Thread.yield(); // lost initialization race; just spinelse if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {//cas操作判断并置为-1try {if ((tab = table) == null || tab.length == 0) {int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//DEFAULT_CAPACITY = 16，若没有参数则大小默认为16@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;}

put操作

    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; K fk; V fv;if (tab == null || (n = tab.length) == 0)//表为空或表长度为0tab = initTable();//初始化表else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//i = (n - 1) & hash为索引值，查找该元素，//如果为null,说明第一次插入if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))break;                   // no lock when adding to empty bin}else if ((fh = f.hash) == MOVED)//MOVED=-1;当前正在扩容，一起进行扩容操作tab = helpTransfer(tab, f);else if (onlyIfAbsent && fh == hash &&  // check first node((fk = f.key) == key || fk != null && key.equals(fk)) &&(fv = f.val) != null)return fv;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);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;}}else if (f instanceof ReservationNode)throw new IllegalStateException("Recursive update");}}if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}addCount(1L, binCount);return null;}//哈希算法static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}//保证拿到最新的数据static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {return (Node<K,V>)U.getObjectAcquire(tab, ((long)i << ASHIFT) + ABASE);}//CAS操作插入节点，比较数组下标为i的节点是否为c，若是，用v交换，否则不操作。//如果CAS成功，表示插入成功，结束循环进行addCount(1L, binCount)看是否需要扩容static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,Node<K,V> c, Node<K,V> v) {return U.compareAndSetObject(tab, ((long)i << ASHIFT) + ABASE, c, v);}

table扩容

当table容量不足的时候，即table的元素数量达到容量阈值sizeCtl，需要对table进行扩容。整个扩容分为两部分：

构建一个nextTable，大小为table的两倍。
把table的数据复制到nextTable中。
这两个过程在单线程下实现很简单，但是ConcurrentHashMap是支持并发插入的，扩容操作自然也会有并发的出现，这种情况下，第二步可以支持节点的并发复制，这样性能自然提升不少，但实现的复杂度也上升了一个台阶。
继续上源码：
第一步，构建nextTable，毫无疑问，这个过程只能只有单个线程进行nextTable的初始化.

private final void addCount(long x, int check) {... 省略部分代码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) {// sc < 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))// 不满足前面5个条件时，尝试参与此次扩容，把正在执行transfer任务的线程数加1，+2代表有1个，+1代表有0个transfer(tab, nt);}//试着让自己成为第一个执行transfer任务的线程else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);// 去执行transfer任务s = sumCount();// 重新计数，判断是否需要开启下一轮扩容}}
}

节点从table移动到nextTable，大体思想是遍历、复制的过程。遍历过所有的节点以后就完成了复制工作，把table指向nextTable，并更新sizeCtl为新数组大小的0.75倍，扩容完成。

get操作

判断table是否为空，如果为空，直接返回null。
计算key的hash值，并获取指定table中指定位置的Node节点，通过遍历链表或则树结构找到对应的节点，返回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;
}

和HashTable的区别：

ConcurrentHashMap 是一个并发散列映射表，它允许完全并发的读取，并且支持给定数量的并发更新。
而HashTable和同步包装器包装的 HashMap，使用一个全局的锁来同步不同线程间的并发访问，同一时间点，只能有一个线程持有锁，也就是说在同一时间点，只能有一个线程能访问容器，这虽然保证多线程间的安全并发访问，但同时也导致对容器的访问变成串行化的了。

总结：

Hashtable的任何操作都会把整个表锁住，是阻塞的。好处是总能获取最实时的更新，比如说线程A调用putAll写入大量数据，期间线程B调用get，线程B就会被阻塞，直到线程A完成putAll，因此线程B肯定能获取到线程A写入的完整数据。坏处是所有调用都要排队，效率较低。
ConcurrentHashMap 是设计为非阻塞的。在更新时会局部锁住某部分数据，但不会把整个表都锁住。同步读取操作则是完全非阻塞的。好处是在保证合理的同步前提下，效率很高。坏处是严格来说读取操作不能保证反映最近的更新。例如线程A调用putAll写入大量数据，期间线程B调用get，则只能get到目前为止已经顺利插入的部分数据。
应该根据具体的应用场景选择合适的HashMap。

转自：https://www.jianshu.com/p/d0b37b927c48

作者：不会游泳的金鱼_

[集合]ConcurrentHashMap的源码分析相关推荐

死磕 java集合之ArrayDeque源码分析
问题 (1)什么是双端队列? (2)ArrayDeque是怎么实现双端队列的? (3)ArrayDeque是线程安全的吗? (4)ArrayDeque是有界的吗? 简介双端队列是一种特殊的队列,它的 ...
【死磕 Java 集合】— LinkedTransferQueue源码分析
[死磕 Java 集合]- LinkedTransferQueue源码分析问题 (1)LinkedTransferQueue是什么东东? (2)LinkedTransferQueue是怎么实现阻塞队 ...
死磕Java集合之BitSet源码分析（JDK18）
死磕Java集合之BitSet源码分析(JDK18) 文章目录死磕Java集合之BitSet源码分析(JDK18) 简介继承体系存储结构源码解析属性构造方法 set(int bitInde ...
java arraydeque_死磕 java集合之ArrayDeque源码分析
问题 (1)什么是双端队列? (2)ArrayDeque是怎么实现双端队列的? (3)ArrayDeque是线程安全的吗? (4)ArrayDeque是有界的吗? 简介双端队列是一种特殊的队列,它的 ...
java ee是什么_死磕 java集合之HashSet源码分析
问题 (1)集合(Collection)和集合(Set)有什么区别? (2)HashSet怎么保证添加元素不重复? (3)HashSet是否允许null元素? (4)HashSet是有序的吗? (5) ...
死磕 java集合之ConcurrentSkipListMap源码分析——发现个bug
前情提要点击链接查看"跳表"详细介绍. 拜托,面试别再问我跳表了! 简介跳表是一个随机化的数据结构,实质就是一种可以进行二分查找的有序链表. 跳表在原有的有序链表上面增加了多级 ...
Java集合：Hashtable源码分析
1. 概述上次讨论了HashMap的结构,原理和实现,本文来对Map家族的另外一个常用集合HashTable进行介绍.HashTable和HashMap两种集合非常相似,经常被各种面试官问到两者的区 ...
java linkedlist源码_Java集合之LinkedList源码分析
一.LinkedList简介 LinkedList是一种可以在任何位置进行高效地插入和移除操作的有序序列,它是基于双向链表实现的. ps:这里有一个问题,就是关于实现LinkedList的数据结构是否 ...
HashMap与ConcurrentHashMap万字源码分析
HashMap与ConcurrentHashMap`源码解析 JDK版本:1.7 & 1.8 开发中常见的数据结构有三种: 1.数组结构:存储区间连续.内存占用严重.空间复杂度大优点:因 ...

[集合]ConcurrentHashMap的源码分析

前言：