关键词:
一、概述
哈希表(hash table)也叫散列表,是一种非常重要的数据结构,应用场景及其丰富,许多缓存技术(比如memcached)的核心其实就是在内存中维护一张大的哈希表。
是根据关键码值(Key value)而直接进行访问的数据结构。也就是说,它通过把关键码值映射到表中一个位置来访问记录,以加快查找的速度。这个映射函数叫做散列函数,存放记录的数组叫做散列表。
给定表M,存在函数f(key),对任意给定的关键字值key,代入函数后若能得到包含该关键字的记录在表中的地址,则称表M为哈希(Hash)表,函数f(key)为哈希(Hash) 函数。
1.1、基本数据结构
数组:采用一段连续的存储单元来存储数据。对于指定下标的查找,时间复杂度为O(1);通过给定值进行查找,需要遍历数组,逐一比对给定关键字和数组元素,时间复杂度为O(n),当然,对于有序数组,则可采用二分查找,插值查找,斐波那契查找等方式,可将查找复杂度提高为O(logn);对于一般的插入删除操作,涉及到数组元素的移动,其平均复杂度也为O(n)
线性链表:对于链表的新增,删除等操作(在找到指定操作位置后),仅需处理结点间的引用即可,时间复杂度为O(1),而查找操作需要遍历链表逐一进行比对,复杂度为O(n)
二叉树:对一棵相对平衡的有序二叉树,对其进行插入,查找,删除等操作,平均复杂度均为O(logn)。
哈希表:相比上述几种数据结构,在哈希表中进行添加,删除,查找等操作,性能十分之高,不考虑哈希冲突的情况下,仅需一次定位即可完成,时间复杂度为O(1)。
二、HashMap
2.1、jdk6-HashMap
JDK6中HashMap 里面是一个数组,然后数组中每个元素是一个单向链表。
每个数组内实体是嵌套类 Entry 的实例,Entry 包含四个属性:key, value, hash 值和用于单向链表的 next。
2.1.1、构造函数
几个构造函数的核心均是以下操作
public HashMap() this.loadFactor = DEFAULT_LOAD_FACTOR; threshold = (int)(DEFAULT_INITIAL_CAPACITY * DEFAULT_LOAD_FACTOR); table = new Entry[DEFAULT_INITIAL_CAPACITY]; init();
capacity:当前数组容量,始终保持 2^n,可以扩容,扩容后数组大小为当前的 2 倍。默认大小16;DEFAULT_INITIAL_CAPACITY = 16;
loadFactor:负载因子,默认为 0.75。
threshold:扩容的阈值,等于 capacity * loadFactor
其中table即是当前hash表,默认大小为16,init默认是一个空实现,为了后续扩展
可以看到table的定义
/** * The table, resized as necessary. Length MUST Always be a power of two. */ transient Entry[] table;
不是线程安全的,并且长度必须是2的倍数
2.1.2、put方法说明
public V put(K key, V value) // 如果 key 为 null,最终会将这个 entry 放到 table[0] 中 if (key == null) return putForNullKey(value); // 1. 求 key 的 hash 值 int hash = hash(key.hashCode()); // 2. 找到对应的数组下标 int i = indexFor(hash, table.length); // 3. 遍历一下对应下标处的链表,看是否有重复的 key 已经存在, // 如果有,直接覆盖,put 方法返回旧值就结束了 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++; // 4. 不存在重复的 key,将此 entry 添加到链表中 addEntry(hash, key, value, i); return null;
1》计算数组位置indexFor
使用 key 的 hash 值对数组长度进行取模
static int indexFor(int h, int length) return h & (length-1);
简单说就是取 hash 值的低 n 位。如在数组长度为 32 的时候,其实取的就是 key 的 hash 值的低 5 位,作为它在数组中的下标位置。
2》添加节点到链表
//将新值放到链表的表头,然后 size++ //参数(hash,key,value,key的hash在hash表中数据位置) void addEntry(int hash, K key, V value, int bucketIndex) Entry<K,V> e = table[bucketIndex]; table[bucketIndex] = new Entry<K,V>(hash, key, value, e); if (size++ >= threshold)// threshold=16*0.75=12 ,即size=12的时候开始扩容 2*16=32 resize(2 * table.length);
这里是先添加,然后扩容threshold=16*0.75=12 ,即size=12的时候开始扩容 2*16=32
3》数组扩容
在插入新值的时候,如果当前的 size 已经达到了阈值,并且要插入的数组位置上已经有元素,那么就会触发扩容,扩容后,数组大小为原来的 2 倍。
void resize(int newCapacity) Entry[] oldTable = table; int oldCapacity = oldTable.length; if (oldCapacity == MAXIMUM_CAPACITY) threshold = Integer.MAX_VALUE; return; // 新的数组 Entry[] newTable = new Entry[newCapacity]; // 将原来数组中的值迁移到新的更大的数组中 transfer(newTable); table = newTable; threshold = (int)(newCapacity * loadFactor);
扩容就是用一个新的大数组替换原来的小数组,并将原来数组中的值迁移到新的数组中。
由于是双倍扩容,迁移过程中,会将原来 table[i] 中的链表的所有节点,分拆到新的数组的 newTable[i] 和 newTable[i + oldLength] 位置上。如原来数组长度是 16,那么扩容后,原来 table[0] 处的链表中的所有元素会被分配到新数组中 newTable[0] 和 newTable[16] 这两个位置。
其中transfer如下
/** * Transfers all entries from current table to newTable. */ void transfer(Entry[] newTable) Entry[] src = table; int newCapacity = newTable.length; for (int j = 0; j < src.length; j++) Entry<K,V> e = src[j]; if (e != null) src[j] = null; do Entry<K,V> next = e.next; int i = indexFor(e.hash, newCapacity); e.next = newTable[i]; newTable[i] = e; e = next; while (e != null);
2.1.3、get方法说明
- 根据 key 计算 hash 值。
- 找到相应的数组下标:hash & (length – 1)。
- 遍历该数组位置处的链表,直到找到相等(==或equals)的 key。
public V get(Object key) //,key 为 null 的话,会被放到 table[0],所以只要遍历下 table[0] 处的链表就可以了 if (key == null) return getForNullKey(); int hash = hash(key.hashCode()); for (Entry<K,V> e = table[indexFor(hash, table.length)]; e != null; e = e.next) Object k; if (e.hash == hash && ((k = e.key) == key || key.equals(k))) return e.value; return null;
即是getEntry也是比较简单
/** * Returns the entry associated with the specified key in the * HashMap. Returns null if the HashMap contains no mapping * for the key. */ final Entry<K,V> getEntry(Object key) int hash = (key == null) ? 0 : hash(key.hashCode()); for (Entry<K,V> e = table[indexFor(hash, table.length)]; e != null; e = e.next) Object k; if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) return e; return null;
2.2、jdk7-HashMap
JDK6中HashMap 里面是一个数组,然后数组中每个元素是一个单向链表。
每个数组内实体是嵌套类 Entry 的实例,Entry 包含四个属性:key, value, hash 值和用于单向链表的 next。
2.2.1、构造函数
几个构造函数的核心均是对以下操作
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();
capacity:当前数组容量,始终保持 2^n,可以扩容,扩容后数组大小为当前的 2 倍。默认大小16;DEFAULT_INITIAL_CAPACITY = 16;
loadFactor:负载因子,默认为 0.75。
threshold:扩容的阈值,默认等于 capacity 即16;与jdk6不同
其中table即是当前hash表,默认大小为16,init默认是一个空实现,为了后续扩展
可以看到Entry[]的table的定义,没有立即被初始化。后续会发现会在 inflateTable 初始化;
2.2.2、put方法说明
public V put(K key, V value) // 当插入第一个元素的时候,需要先初始化数组大小 if (table == EMPTY_TABLE) inflateTable(threshold); // 如果 key 为 null,感兴趣的可以往里看,最终会将这个 entry 放到 table[0] 中 if (key == null) return putForNullKey(value); // 1. 求 key 的 hash 值 int hash = hash(key); // 2. 找到对应的数组下标 int i = indexFor(hash, table.length); // 3. 遍历一下对应下标处的链表,看是否有重复的 key 已经存在, // 如果有,直接覆盖,put 方法返回旧值就结束了 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++; // 4. 不存在重复的 key,将此 entry 添加到链表中,细节后面说 addEntry(hash, key, value, i); return null;
1》hash表初始化inflateTable
private void inflateTable(int toSize) // Find a power of 2 >= toSize // 保证数组大小一定是 2 的 n 次方。 // 比如这样初始化:new HashMap(20),那么处理成初始数组大小是 32 int capacity = roundUpToPowerOf2(toSize); // 计算扩容阈值:capacity * loadFactor threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1); //初始化数组 table = new Entry[capacity]; initHashSeedAsNeeded(capacity);
初始threshold=initialCapacity=16,这是重新计算threshold=12
2》计算数组位置indexFor【同jdk6】
3》添加节点到链表
找到数组下标后,会先进行 key 判重,如果没有重复,就准备将新值放入到链表的表头。
void addEntry(int hash, K key, V value, int bucketIndex) // 如果当前 HashMap 大小已经达到了阈值,并且新值要插入的数组位置已经有元素了,那么要扩容 if ((size >= threshold) && (null != table[bucketIndex])) // 扩容,后面会介绍一下 resize(2 * table.length); // 扩容以后,重新计算 hash 值 hash = (null != key) ? hash(key) : 0; // 重新计算扩容后的新的下标 bucketIndex = indexFor(hash, table.length); createEntry(hash, key, value, bucketIndex); // 就是将新值放到链表的表头,然后 size++ void createEntry(int hash, K key, V value, int bucketIndex) Entry<K,V> e = table[bucketIndex]; table[bucketIndex] = new Entry<>(hash, key, value, e); size++;
先判断是否需要扩容,需要的话先扩容,然后再将这个新的数据插入到扩容后的数组的相应位置处的链表的表头。
4》数组扩容
在插入新值的时候,如果当前的 size 已经达到了阈值,并且要插入的数组位置上已经有元素,那么就会触发扩容,扩容后,数组大小为原来的 2 倍。
void resize(int newCapacity) Entry[] oldTable = table; int oldCapacity = oldTable.length; if (oldCapacity == MAXIMUM_CAPACITY) threshold = Integer.MAX_VALUE; return; // 新的数组 Entry[] newTable = new Entry[newCapacity]; // 将原来数组中的值迁移到新的更大的数组中 transfer(newTable, initHashSeedAsNeeded(newCapacity)); table = newTable; threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1);
只是最后3行与6有所差异,但功能一致。
扩容就是用一个新的大数组替换原来的小数组,并将原来数组中的值迁移到新的数组中。
由于是双倍扩容,迁移过程中,会将原来 table[i] 中的链表的所有节点,分拆到新的数组的 newTable[i] 和 newTable[i + oldLength] 位置上。如原来数组长度是 16,那么扩容后,原来 table[0] 处的链表中的所有元素会被分配到新数组中 newTable[0] 和 newTable[16] 这两个位置。
其中transfer如下
/** * Transfers all entries from current table to newTable. */ void transfer(Entry[] newTable, boolean rehash) int newCapacity = newTable.length; for (Entry<K,V> e : table) 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;
2.2.3、get方法说明
- 根据 key 计算 hash 值。
- 找到相应的数组下标:hash & (length – 1)。
- 遍历该数组位置处的链表,直到找到相等(==或equals)的 key。
public V get(Object key) // key 为 null 的话,会被放到 table[0],所以只要遍历下 table[0] 处的链表 if (key == null) return getForNullKey(); // Entry<K,V> entry = getEntry(key); return null == entry ? null : entry.getValue();
即是getEntry也是比较简单
final Entry<K,V> getEntry(Object key) if (size == 0) return null; int hash = (key == null) ? 0 : hash(key); // 确定数组下标,然后从头开始遍历链表,直到找到为止 for (Entry<K,V> e = table[indexFor(hash, table.length)]; e != null; e = e.next) Object k; if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) return e; return null;
2.3、jdk8-HashMap
Java8 对 HashMap 进行了一些修改,最大的不同就是利用了红黑树,所以其由 数组+链表+红黑树 组成。
根据 Java7 HashMap 的介绍,我们知道,查找的时候,根据 hash 值我们能够快速定位到数组的具体下标,但是之后的话,需要顺着链表一个个比较下去才能找到我们需要的,时间复杂度取决于链表的长度,为 O(n)。
为了降低这部分的开销,在 Java8 中,当链表中的元素超过了 8 个以后,会将链表转换为红黑树,在这些位置进行查找的时候可以降低时间复杂度为 O(logN)。
Java7 中使用 Entry 来代表每个 HashMap 中的数据节点,Java8 中使用 Node,基本没有区别,都是 key,value,hash 和 next 这四个属性,不过,Node 只能用于链表的情况,红黑树的情况需要使用 TreeNode。
我们根据数组元素中,第一个节点数据类型是 Node 还是 TreeNode 来判断该位置下是链表还是红黑树的。
2.3.1、构造函数
几个构造函数的核心均是将以下参数赋值
capacity:当前数组容量,始终保持 2^n,可以扩容,扩容后数组大小为当前的 2 倍。默认大小16;DEFAULT_INITIAL_CAPACITY = 16;
loadFactor:负载因子,默认为 0.75。
threshold:扩容的阈值,默认等于 capacity
其中table即是当前hash表,默认大小为16,init默认是一个空实现,为了后续扩展
可以看到Node的table的定义
/** * The table, initialized on first use, and resized as * necessary. When allocated, length is always a power of two. * (We also tolerate length zero in some operations to allow * bootstrapping mechanics that are currently not needed.) */ transient Node<K,V>[] table;
查看Node如下
/** * Basic hash bin node, used for most entries. (See below for * TreeNode subclass, and in LinkedHashMap for its Entry subclass.) */ static class Node<K,V> implements Map.Entry<K,V> final int hash; final K key; V value; Node<K,V> next; Node(int hash, K key, V value, Node<K,V> next) this.hash = hash; this.key = key; this.value = value; this.next = next; public final K getKey() return key; public final V getValue() return value; public final String toString() return key + "=" + value; public final int hashCode() return Objects.hashCode(key) ^ Objects.hashCode(value); public final V setValue(V newValue) V oldValue = value; value = newValue; return oldValue; public final boolean equals(Object o) if (o == this) return true; if (o instanceof Map.Entry) Map.Entry<?,?> e = (Map.Entry<?,?>)o; if (Objects.equals(key, e.getKey()) && Objects.equals(value, e.getValue())) return true; return false;
2.3.2、put方法说明
public V put(K key, V value) return putVal(hash(key), key, value, false, true); // 第三个参数 onlyIfAbsent 如果是 true,那么只有在不存在该 key 时才会进行 put 操作 // 第四个参数 evict 我们这里不关心 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 值的时候,会触发下面的 resize(),类似 java7 的第一次 put 也要初始化数组长度 // 第一次 resize 和后续的扩容有些不一样,因为这次是数组从 null 初始化到默认的 16 或自定义的初始容量 if ((tab = table) == null || (n = tab.length) == 0) n = (tab = resize()).length; // 找到具体的数组下标,如果此位置没有值,那么直接初始化一下 Node 并放置在这个位置就可以了 if ((p = tab[i = (n - 1) & hash]) == null) tab[i] = newNode(hash, key, value, null); else // 数组该位置有数据 Node<K,V> e; K k; // 首先,判断该位置的第一个数据和我们要插入的数据,key 是不是"相等",如果是,取出这个节点 if (p.hash == hash && ((k = p.key) == key || (key != null && key.equals(k)))) e = p; // 如果该节点是代表红黑树的节点,调用红黑树的插值方法,不展开说红黑树 else if (p instanceof TreeNode) e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value); else // 到这里,说明数组该位置上是一个链表 for (int binCount = 0; ; ++binCount) // 插入到链表的最后面(Java7 是插入到链表的最前面) if ((e = p.next) == null) p.next = newNode(hash, key, value, null); // TREEIFY_THRESHOLD 为 8,所以,如果新插入的值是链表中的第 9 个 // 会触发下面的 treeifyBin,也就是将链表转换为红黑树 if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st treeifyBin(tab, hash); break; // 如果在该链表中找到了"相等"的 key(== 或 equals) if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) // 此时 break,那么 e 为链表中[与要插入的新值的 key "相等"]的 node break; p = e; // e!=null 说明存在旧值的key与要插入的key"相等" // 对于我们分析的put操作,下面这个 if 其实就是进行 "值覆盖",然后返回旧值 if (e != null) V oldValue = e.value; if (!onlyIfAbsent || oldValue == null) e.value = value; afterNodeAccess(e); return oldValue; ++modCount; // 如果 HashMap 由于新插入这个值导致 size 已经超过了阈值,需要进行扩容 if (++size > threshold) resize(); afterNodeInsertion(evict); return null;
和 Java7 稍微有点不一样的地方就是,Java7 是先扩容后插入新值的,Java8 先插值再扩容
1》数组扩容
resize() 方法用于初始化数组或数组扩容,每次扩容后,容量为原来的 2 倍,并进行数据迁移。
final Node<K,V>[] resize() Node<K,V>[] oldTab = table; int oldCap = (oldTab == null) ? 0 : oldTab.length; int oldThr = threshold; int newCap, newThr = 0; if (oldCap > 0) // 对应数组扩容 if (oldCap >= MAXIMUM_CAPACITY) threshold = Integer.MAX_VALUE; return oldTab; // 将数组大小扩大一倍 else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY && oldCap >= DEFAULT_INITIAL_CAPACITY) // 将阈值扩大一倍 newThr = oldThr << 1; // double threshold else if (oldThr > 0) // 对应使用 new HashMap(int initialCapacity) 初始化后,第一次 put 的时候 newCap = oldThr; else // 对应使用 new HashMap() 初始化后,第一次 put 的时候 newCap = DEFAULT_INITIAL_CAPACITY; newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY); if (newThr == 0) float ft = (float)newCap * loadFactor; newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ? (int)ft : Integer.MAX_VALUE); threshold = newThr; // 用新的数组大小初始化新的数组 Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap]; table = newTab; // 如果是初始化数组,到这里就结束了,返回 newTab 即可 if (oldTab != null) // 开始遍历原数组,进行数据迁移。 for (int j = 0; j < oldCap; ++j) Node<K,V> e; if ((e = oldTab[j]) != null) oldTab[j] = null; // 如果该数组位置上只有单个元素,那就简单了,简单迁移这个元素就可以了 if (e.next == null) newTab[e.hash & (newCap - 1)] = e; // 如果是红黑树,具体我们就不展开了 else if (e instanceof TreeNode) ((TreeNode<K,V>)e).split(this, newTab, j, oldCap); else // 这块是处理链表的情况, // 需要将此链表拆成两个链表,放到新的数组中,并且保留原来的先后顺序 // loHead、loTail 对应一条链表,hiHead、hiTail 对应另一条链表,代码还是比较简单的 Node<K,V> loHead = null, loTail = null; Node<K,V> hiHead = null, hiTail = null; Node<K,V> next; do next = e.next; if ((e.hash & oldCap) == 0) if (loTail == null) loHead = e; else loTail.next = e; loTail = e; else if (hiTail == null) hiHead = e; else hiTail.next = e; hiTail = e; while ((e = next) != null); if (loTail != null) loTail.next = null; // 第一条链表 newTab[j] = loHead; if (hiTail != null) hiTail.next = null; // 第二条链表的新的位置是 j + oldCap,这个很好理解 newTab[j + oldCap] = hiHead; return newTab;
2.3.3、get方法说明
- 计算 key 的 hash 值,根据 hash 值找到对应数组下标: hash & (length-1)
- 判断数组该位置处的元素是否刚好就是我们要找的,如果不是,走第三步
- 判断该元素类型是否是 TreeNode,如果是,用红黑树的方法取数据,如果不是,走第四步
- 遍历链表,直到找到相等(==或equals)的 key
public V get(Object key) Node<K,V> e; return (e = getNode(hash(key), key)) == null ? null : e.value; final Node<K,V> getNode(int hash, Object key) Node<K,V>[] tab; Node<K,V> first, e; int n; K k; if ((tab = table) != null && (n = tab.length) > 0 && (first = tab[(n - 1) & hash]) != null) // 判断第一个节点是不是就是需要的 if (first.hash == hash && // always check first node ((k = first.key) == key || (key != null && key.equals(k)))) return first; if ((e = first.next) != null) // 判断是否是红黑树 if (first instanceof TreeNode) return ((TreeNode<K,V>)first).getTreeNode(hash, key); // 链表遍历 do if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) return e; while ((e = e.next) != null); return null;
三、ConcurrentHashMap
3.1、jdk6-ConcurrentHashMap
整个 ConcurrentHashMap 由一个个 Segment 组成,Segment 代表”部分“或”一段“的意思,所以很多地方都会将其描述为分段锁用了“槽”来代表一个 segment。
简单理解就是,ConcurrentHashMap 是一个 Segment 数组,Segment 通过继承 ReentrantLock 来进行加锁,所以每次需要加锁的操作锁住的是一个 segment,这样只要保证每个 Segment 是线程安全的,也就实现了全局的线程安全。
3.1.1、构造函数
几个构造函数的核心均是以下操作
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; // 计算并行级别 ssize,因为要保持并行级别是 2 的 n 次方 while (ssize < concurrencyLevel) ++sshift; ssize <<= 1; // 用默认值,concurrencyLevel 为 16,sshift 为 4 // 那么计算出 segmentShift 为 28,segmentMask 为 15,后面会用到这两个值 segmentShift = 32 - sshift; segmentMask = ssize - 1; this.segments = Segment.newArray(ssize); if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; // initialCapacity 是设置整个 map 初始的大小, // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小 // 如 initialCapacity 为 64,那么每个 Segment 或称之为"槽"可以分到 4 个 int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; int cap = 1; while (cap < c) cap <<= 1; for (int i = 0; i < this.segments.length; ++i) this.segments[i] = new Segment<K,V>(cap, loadFactor);
concurrencyLevel:并行级别、并发数、Segment 数,默认是 16,也就是说 ConcurrentHashMap 有 16 个 Segments,所以理论上,这个时候,最多可以同时支持 16 个线程并发写,只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值,但是一旦初始化以后,它是不可以扩容的。
再具体到每个 Segment 内部,其实每个 Segment 很像之前介绍的 HashMap
initialCapacity:初始容量,这个值指的是整个 ConcurrentHashMap 的初始容量,实际操作的时候需要平均分给每个 Segment。
loadFactor:负载因子,之前我们说了,Segment 数组不可以扩容,所以这个负载因子是给每个 Segment 内部使用的。
可以看到原table,现在segments的定义
/** * The segments, each of which is a specialized hash table */ final Segment<K,V>[] segments;
其中Segments定义如下
/** * Segments are specialized versions of hash tables. This * subclasses from ReentrantLock opportunistically, just to * simplify some locking and avoid separate construction. */ static final class Segment<K,V> extends ReentrantLock implements Serializable /* * Segments maintain a table of entry lists that are ALWAYS * kept in a consistent state, so can be read without locking. * Next fields of nodes are immutable (final). All list * additions are performed at the front of each bin. This * makes it easy to check changes, and also fast to traverse. * When nodes would otherwise be changed, new nodes are * created to replace them. This works well for hash tables * since the bin lists tend to be short. (The average length * is less than two for the default load factor threshold.) * * Read operations can thus proceed without locking, but rely * on selected uses of volatiles to ensure that completed * write operations performed by other threads are * noticed. For most purposes, the "count" field, tracking the * number of elements, serves as that volatile variable * ensuring visibility. This is convenient because this field * needs to be read in many read operations anyway: * * - All (unsynchronized) read operations must first read the * "count" field, and should not look at table entries if * it is 0. * * - All (synchronized) write operations should write to * the "count" field after structurally changing any bin. * The operations must not take any action that could even * momentarily cause a concurrent read operation to see * inconsistent data. This is made easier by the nature of * the read operations in Map. For example, no operation * can reveal that the table has grown but the threshold * has not yet been updated, so there are no atomicity * requirements for this with respect to reads. * * As a guide, all critical volatile reads and writes to the * count field are marked in code comments. */ private static final long serialVersionUID = 2249069246763182397L; /** * The number of elements in this segment‘s region. */ transient volatile int count; /** * Number of updates that alter the size of the table. This is * used during bulk-read methods to make sure they see a * consistent snapshot: If modCounts change during a traversal * of segments computing size or checking containsValue, then * we might have an inconsistent view of state so (usually) * must retry. */ transient int modCount; /** * The table is rehashed when its size exceeds this threshold. * (The value of this field is always <tt>(int)(capacity * * loadFactor)</tt>.) */ transient int threshold; /** * The per-segment table. */ transient volatile HashEntry<K,V>[] table; /** * The load factor for the 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; Segment(int initialCapacity, float lf) loadFactor = lf; setTable(HashEntry.<K,V>newArray(initialCapacity)); @SuppressWarnings("unchecked") static final <K,V> Segment<K,V>[] newArray(int i) return new Segment[i]; /** * Sets table to new HashEntry array. * Call only while holding lock or in constructor. */ void setTable(HashEntry<K,V>[] newTable) threshold = (int)(newTable.length * loadFactor); table = newTable; /** * Returns properly casted first entry of bin for given hash. */ HashEntry<K,V> getFirst(int hash) HashEntry<K,V>[] tab = table; return tab[hash & (tab.length - 1)]; /** * Reads value field of an entry under lock. Called if value * field ever appears to be null. This is possible only if a * compiler happens to reorder a HashEntry initialization with * its table assignment, which is legal under memory model * but is not known to ever occur. */ V readValueUnderLock(HashEntry<K,V> e) lock(); try return e.value; finally unlock(); /* Specialized implementations of map methods */ V get(Object key, int hash) if (count != 0) // read-volatile HashEntry<K,V> e = getFirst(hash); while (e != null) if (e.hash == hash && key.equals(e.key)) V v = e.value; if (v != null) return v; return readValueUnderLock(e); // recheck e = e.next; return null; boolean containsKey(Object key, int hash) if (count != 0) // read-volatile HashEntry<K,V> e = getFirst(hash); while (e != null) if (e.hash == hash && key.equals(e.key)) return true; e = e.next; return false; boolean containsValue(Object value) if (count != 0) // read-volatile HashEntry<K,V>[] tab = table; int len = tab.length; for (int i = 0 ; i < len; i++) for (HashEntry<K,V> e = tab[i]; e != null; e = e.next) V v = e.value; if (v == null) // recheck v = readValueUnderLock(e); if (value.equals(v)) return true; return false; boolean replace(K key, int hash, V oldValue, V newValue) lock(); try HashEntry<K,V> e = getFirst(hash); while (e != null && (e.hash != hash || !key.equals(e.key))) e = e.next; boolean replaced = false; if (e != null && oldValue.equals(e.value)) replaced = true; e.value = newValue; return replaced; finally unlock(); V replace(K key, int hash, V newValue) lock(); try HashEntry<K,V> e = getFirst(hash); while (e != null && (e.hash != hash || !key.equals(e.key))) e = e.next; V oldValue = null; if (e != null) oldValue = e.value; e.value = newValue; return oldValue; finally unlock(); V put(K key, int hash, V value, boolean onlyIfAbsent) lock(); try int c = count; if (c++ > threshold) // ensure capacity rehash(); HashEntry<K,V>[] tab = table; int index = hash & (tab.length - 1); HashEntry<K,V> first = tab[index]; HashEntry<K,V> e = first; while (e != null && (e.hash != hash || !key.equals(e.key))) e = e.next; V oldValue; if (e != null) oldValue = e.value; if (!onlyIfAbsent) e.value = value; else oldValue = null; ++modCount; tab[index] = new HashEntry<K,V>(key, hash, first, value); count = c; // write-volatile return oldValue; finally unlock(); void rehash() HashEntry<K,V>[] oldTable = table; int oldCapacity = oldTable.length; if (oldCapacity >= MAXIMUM_CAPACITY) return; /* * Reclassify nodes in each list to new Map. Because we are * using power-of-two expansion, the elements from each bin * must either stay at same index, or move with a power of two * offset. We eliminate unnecessary node creation by catching * cases where old nodes can be reused because their next * fields won‘t change. Statistically, at the default * threshold, only about one-sixth of them need cloning when * a table doubles. The nodes they replace will be garbage * collectable as soon as they are no longer referenced by any * reader thread that may be in the midst of traversing table * right now. */ HashEntry<K,V>[] newTable = HashEntry.newArray(oldCapacity<<1); threshold = (int)(newTable.length * loadFactor); int sizeMask = newTable.length - 1; for (int i = 0; i < oldCapacity ; i++) // We need to guarantee that any existing reads of old Map can // proceed. So we cannot yet null out each bin. HashEntry<K,V> e = oldTable[i]; if (e != null) HashEntry<K,V> next = e.next; int idx = e.hash & sizeMask; // Single node on list if (next == null) newTable[idx] = e; else // Reuse trailing consecutive sequence at same 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 all remaining nodes for (HashEntry<K,V> p = e; p != lastRun; p = p.next) int k = p.hash & sizeMask; HashEntry<K,V> n = newTable[k]; newTable[k] = new HashEntry<K,V>(p.key, p.hash, n, p.value); table = newTable; /** * Remove; match on key only if value null, else match both. */ V remove(Object key, int hash, Object value) lock(); try int c = count - 1; HashEntry<K,V>[] tab = table; int index = hash & (tab.length - 1); HashEntry<K,V> first = tab[index]; HashEntry<K,V> e = first; while (e != null && (e.hash != hash || !key.equals(e.key))) e = e.next; V oldValue = null; if (e != null) V v = e.value; if (value == null || value.equals(v)) oldValue = v; // All entries following removed node can stay // in list, but all preceding ones need to be // cloned. ++modCount; HashEntry<K,V> newFirst = e.next; for (HashEntry<K,V> p = first; p != e; p = p.next) newFirst = new HashEntry<K,V>(p.key, p.hash, newFirst, p.value); tab[index] = newFirst; count = c; // write-volatile return oldValue; finally unlock(); void clear() if (count != 0) lock(); try HashEntry<K,V>[] tab = table; for (int i = 0; i < tab.length ; i++) tab[i] = null; ++modCount; count = 0; // write-volatile finally unlock();
是线程安全的,其通过ReentrantLock 实现
3.1.2、put方法说明
public V put(K key, V value) if (value == null) throw new NullPointerException(); int hash = hash(key.hashCode()); return segmentFor(hash).put(key, hash, value, false);
1》计算当前key的hash的segment对象segmentFor
final Segment<K,V> segmentFor(int hash) return segments[(hash >>> segmentShift) & segmentMask];
2》使用segment将key和value到table数组
V put(K key, int hash, V value, boolean onlyIfAbsent) lock(); try int c = count; if (c++ > threshold) // ensure capacity rehash(); HashEntry<K,V>[] tab = table; // 再利用 hash 值,求应该放置的数组下标 int index = hash & (tab.length - 1); // first 是数组该位置处的链表的表头 HashEntry<K,V> first = tab[index]; HashEntry<K,V> e = first; while (e != null && (e.hash != hash || !key.equals(e.key))) e = e.next; V oldValue; if (e != null) oldValue = e.value; if (!onlyIfAbsent) e.value = value; else oldValue = null; ++modCount; tab[index] = new HashEntry<K,V>(key, hash, first, value); count = c; // write-volatile return oldValue; finally unlock();
3.1.3、get方法说明
- 计算 hash 值,找到 segment 数组中的具体位置,或我们前面用的“槽”
- 槽中也是一个数组,根据 hash 找到数组中具体的位置
- 到这里是链表了,顺着链表进行查找即可
public V get(Object key) int hash = hash(key.hashCode()); return segmentFor(hash).get(key, hash);
V get(Object key, int hash) if (count != 0) // read-volatile HashEntry<K,V> e = getFirst(hash); while (e != null) if (e.hash == hash && key.equals(e.key)) V v = e.value; if (v != null) return v; return readValueUnderLock(e); // recheck e = e.next; return null;
3.2、jdk7-ConcurrentHashMap
整个 ConcurrentHashMap 由一个个 Segment 组成,Segment 代表”部分“或”一段“的意思,所以很多地方都会将其描述为分段锁用了“槽”来代表一个 segment。
简单理解就是,ConcurrentHashMap 是一个 Segment 数组,Segment 通过继承 ReentrantLock 来进行加锁,所以每次需要加锁的操作锁住的是一个 segment,这样只要保证每个 Segment 是线程安全的,也就实现了全局的线程安全。
3.2.1、构造函数
几个构造函数的核心均是以下操作
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; // 计算并行级别 ssize,因为要保持并行级别是 2 的 n 次方 while (ssize < concurrencyLevel) ++sshift; ssize <<= 1; // 我们这里先不要那么烧脑,用默认值,concurrencyLevel 为 16,sshift 为 4 // 那么计算出 segmentShift 为 28,segmentMask 为 15,后面会用到这两个值 this.segmentShift = 32 - sshift; this.segmentMask = ssize - 1; if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; // initialCapacity 是设置整个 map 初始的大小, // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小 // 如 initialCapacity 为 64,那么每个 Segment 或称之为"槽"可以分到 4 个 int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2,这个值也是有讲究的,因为这样的话,对于具体的槽上, // 插入一个元素不至于扩容,插入第二个的时候才会扩容 int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c) cap <<= 1; // 创建 Segment 数组, // 并创建数组的第一个元素 segment[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]; // 往数组写入 segment[0] UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] this.segments = ss;
concurrencyLevel:并行级别、并发数、Segment 数,默认是 16,也就是说 ConcurrentHashMap 有 16 个 Segments,所以理论上,这个时候,最多可以同时支持 16 个线程并发写,只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值,但是一旦初始化以后,它是不可以扩容的。
再具体到每个 Segment 内部,其实每个 Segment 很像之前介绍的 HashMap
initialCapacity:初始容量,这个值指的是整个 ConcurrentHashMap 的初始容量,实际操作的时候需要平均分给每个 Segment。
loadFactor:负载因子,之前我们说了,Segment 数组不可以扩容,所以这个负载因子是给每个 Segment 内部使用的。
初始化完成,得到了一个 Segment 数组。
用 new ConcurrentHashMap() 无参构造函数进行初始化的,那么初始化完成后:
- Segment 数组长度为 16,不可以扩容
- Segment[i] 的默认大小为 2,负载因子是 0.75,得出初始阈值为 1.5,也就是以后插入第一个元素不会触发扩容,插入第二个会进行第一次扩容
- 这里初始化了 segment[0],其他位置还是 null,至于为什么要初始化 segment[0],后面的代码会介绍
- 当前 segmentShift 的值为 32 – 4 = 28,segmentMask 为 16 – 1 = 15,姑且把它们简单翻译为移位数和掩码,这两个值马上就会用到
可以看到原table,现在segments的定义
/** * The segments, each of which is a specialized hash table */ final Segment<K,V>[] segments;
其中Segments定义如下
/** * Segments are specialized versions of hash tables. This * subclasses from ReentrantLock opportunistically, just to * simplify some locking and avoid separate construction. */ static final class Segment<K,V> extends ReentrantLock implements Serializable /* * Segments maintain a table of entry lists that are always * kept in a consistent state, so can be read (via volatile * reads of segments and tables) without locking. This * requires replicating nodes when necessary during table * resizing, so the old lists can be traversed by readers * still using old version of table. * * This class defines only mutative methods requiring locking. * Except as noted, the methods of this class perform the * per-segment versions of ConcurrentHashMap methods. (Other * methods are integrated directly into ConcurrentHashMap * methods.) These mutative methods use a form of controlled * spinning on contention via methods scanAndLock and * scanAndLockForPut. These intersperse tryLocks with * traversals to locate nodes. The main benefit is to absorb * cache misses (which are very common for hash tables) while * obtaining locks so that traversal is faster once * acquired. We do not actually use the found nodes since they * must be re-acquired under lock anyway to ensure sequential * consistency of updates (and in any case may be undetectably * stale), but they will normally be much faster to re-locate. * Also, scanAndLockForPut speculatively creates a fresh node * to use in put if no node is found. */ private static final long serialVersionUID = 2249069246763182397L; /** * The maximum number of times to tryLock in a prescan before * possibly blocking on acquire in preparation for a locked * segment operation. On multiprocessors, using a bounded * number of retries maintains cache acquired while locating * nodes. */ static final int MAX_SCAN_RETRIES = Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1; /** * The per-segment table. Elements are accessed via * entryAt/setEntryAt providing volatile semantics. */ transient volatile HashEntry<K,V>[] table; /** * The number of elements. Accessed only either within locks * or among other volatile reads that maintain visibility. */ transient int count; /** * The total number of mutative operations in this segment. * Even though this may overflows 32 bits, it provides * sufficient accuracy for stability checks in CHM isEmpty() * and size() methods. Accessed only either within locks or * among other volatile reads that maintain visibility. */ transient int modCount; /** * The table is rehashed when its size exceeds this threshold. * (The value of this field is always <tt>(int)(capacity * * loadFactor)</tt>.) */ transient int threshold; /** * The load factor for the 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; Segment(float lf, int threshold, HashEntry<K,V>[] tab) this.loadFactor = lf; this.threshold = threshold; this.table = tab; 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); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; finally unlock(); return oldValue; /** * Doubles size of table and repacks entries, also adding the * given node to new table */ @SuppressWarnings("unchecked") private void rehash(HashEntry<K,V> node) /* * Reclassify nodes in each list to new table. Because we * are using power-of-two expansion, the elements from * each bin must either stay at same index, or move with a * power of two offset. We eliminate unnecessary node * creation by catching cases where old nodes can be * reused because their next fields won‘t change. * Statistically, at the default threshold, only about * one-sixth of them need cloning when a table * doubles. The nodes they replace will be garbage * collectable as soon as they are no longer referenced by * any reader thread that may be in the midst of * concurrently traversing table. Entry accesses use plain * array indexing because they are followed by volatile * table write. */ 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) // Single node on list newTable[idx] = e; else // Reuse consecutive sequence at same 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; /** * Scans for a node containing given key while trying to * acquire lock, creating and returning one if not found. Upon * return, guarantees that lock is held. UNlike in most * methods, calls to method equals are not screened: Since * traversal speed doesn‘t matter, we might as well help warm * up the associated code and accesses as well. * * @return a new node if key not found, else null */ private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; // negative while locating node while (!tryLock()) HashEntry<K,V> f; // to recheck first below if (retries < 0) if (e == null) if (node == null) // speculatively create node node = new HashEntry<K,V>(hash, key, value, null); retries = 0; else if (key.equals(e.key)) retries = 0; else e = e.next; else if (++retries > MAX_SCAN_RETRIES) lock(); break; else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) e = first = f; // re-traverse if entry changed retries = -1; return node; /** * Scans for a node containing the given key while trying to * acquire lock for a remove or replace operation. Upon * return, guarantees that lock is held. Note that we must * lock even if the key is not found, to ensure sequential * consistency of updates. */ private void scanAndLock(Object key, int hash) // similar to but simpler than scanAndLockForPut HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; int retries = -1; while (!tryLock()) HashEntry<K,V> f; if (retries < 0) if (e == null || key.equals(e.key)) retries = 0; else e = e.next; else if (++retries > MAX_SCAN_RETRIES) lock(); break; else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) e = first = f; retries = -1; /** * Remove; match on key only if value null, else match both. */ final V remove(Object key, int hash, Object value) if (!tryLock()) scanAndLock(key, hash); V oldValue = null; try HashEntry<K,V>[] tab = table; int index = (tab.length - 1) & hash; HashEntry<K,V> e = entryAt(tab, index); HashEntry<K,V> pred = null; while (e != null) K k; HashEntry<K,V> next = e.next; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) V v = e.value; if (value == null || value == v || value.equals(v)) if (pred == null) setEntryAt(tab, index, next); else pred.setNext(next); ++modCount; --count; oldValue = v; break; pred = e; e = next; finally unlock(); return oldValue; final boolean replace(K key, int hash, V oldValue, V newValue) if (!tryLock()) scanAndLock(key, hash); boolean replaced = false; try HashEntry<K,V> e; for (e = entryForHash(this, hash); e != null; e = e.next) K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) if (oldValue.equals(e.value)) e.value = newValue; ++modCount; replaced = true; break; finally unlock(); return replaced; final V replace(K key, int hash, V value) if (!tryLock()) scanAndLock(key, hash); V oldValue = null; try HashEntry<K,V> e; for (e = entryForHash(this, hash); e != null; e = e.next) K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) oldValue = e.value; e.value = value; ++modCount; break; finally unlock(); return oldValue; final void clear() lock(); try HashEntry<K,V>[] tab = table; for (int i = 0; i < tab.length ; i++) setEntryAt(tab, i, null); ++modCount; count = 0; finally unlock();
是线程安全的,其通过ReentrantLock 实现
3.2.2、put方法说明
public V put(K key, V value) Segment<K,V> s; if (value == null) throw new NullPointerException(); // 1. 计算 key 的 hash 值 int hash = hash(key); // 2. 根据 hash 值找到 Segment 数组中的位置 j // hash 是 32 位,无符号右移 segmentShift(28) 位,剩下低 4 位, // 然后和 segmentMask(15) 做一次与操作,也就是说 j 是 hash 值的最后 4 位,也就是槽的数组下标 int j = (hash >>> segmentShift) & segmentMask; // 刚刚说了,初始化的时候初始化了 segment[0],但是其他位置还是 null, // ensureSegment(j) 对 segment[j] 进行初始化 if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); // 3. 插入新值到 槽 s 中 return s.put(key, hash, value, false);
Segment 内部是由 数组+链表 组成的。这对s.put
final V put(K key, int hash, V value, boolean onlyIfAbsent) // 在往该 segment 写入前,需要先获取该 segment 的独占锁 // 先看主流程,后面还会具体介绍这部分内容 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try // 这个是 segment 内部的数组 HashEntry<K,V>[] tab = table; // 再利用 hash 值,求应该放置的数组下标 int index = (tab.length - 1) & hash; // first 是数组该位置处的链表的表头 HashEntry<K,V> first = entryAt(tab, index); // 下面这串 for 循环虽然很长,不过也很好理解,想想该位置没有任何元素和已经存在一个链表这两种情况 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 // node 到底是不是 null,这个要看获取锁的过程,不过和这里都没有关系。 // 如果不为 null,那就直接将它设置为链表表头;如果是null,初始化并设置为链表表头。 if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; // 如果超过了该 segment 的阈值,这个 segment 需要扩容 if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); // 扩容后面也会具体分析 else // 没有达到阈值,将 node 放到数组 tab 的 index 位置, // 其实就是将新的节点设置成原链表的表头 setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; finally // 解锁 unlock(); return oldValue;
1》初始化槽: ensureSegment
ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0],对于其他槽来说,在插入第一个值的时候进行初始化。
这里需要考虑并发,因为很可能会有多个线程同时进来初始化同一个槽 segment[k],不过只要有一个成功了就可以。
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; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) // 这里看到为什么之前要初始化 segment[0] 了, // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k] // 为什么要用“当前”,因为 segment[0] 可能早就扩容过了 Segment<K,V> proto = ss[0]; int cap = proto.table.length; float lf = proto.loadFactor; int threshold = (int)(cap * lf); // 初始化 segment[k] 内部的数组 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) // 再次检查一遍该槽是否被其他线程初始化了。 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); // 使用 while 循环,内部用 CAS,当前线程成功设值或其他线程成功设值后,退出 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; return seg;
ensureSegment(int k) 比较简单,对于并发操作使用 CAS 进行控制。
2》获取写入锁: scanAndLockForPut
前面我们看到,在往某个 segment 中 put 的时候,首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value),也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁,如果失败,那么进入到 scanAndLockForPut 这个方法来获取锁。
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; // negative while locating node // 循环获取锁 while (!tryLock()) HashEntry<K,V> f; // to recheck first below if (retries < 0) if (e == null) if (node == null) // speculatively create node // 进到这里说明数组该位置的链表是空的,没有任何元素 // 当然,进到这里的另一个原因是 tryLock() 失败,所以该槽存在并发,不一定是该位置 node = new HashEntry<K,V>(hash, key, value, null); retries = 0; else if (key.equals(e.key)) retries = 0; else // 顺着链表往下走 e = e.next; // 重试次数如果超过 MAX_SCAN_RETRIES(单核1多核64),那么不抢了,进入到阻塞队列等待锁 // lock() 是阻塞方法,直到获取锁后返回 else if (++retries > MAX_SCAN_RETRIES) lock(); break; else if ((retries & 1) == 0 && // 这个时候是有大问题了,那就是有新的元素进到了链表,成为了新的表头 // 所以这边的策略是,相当于重新走一遍这个 scanAndLockForPut 方法 (f = entryForHash(this, hash)) != first) e = first = f; // re-traverse if entry changed retries = -1; return node;
这个方法有两个出口,一个是 tryLock() 成功了,循环终止,另一个就是重试次数超过了 MAX_SCAN_RETRIES,进到 lock() 方法,此方法会阻塞等待,直到成功拿到独占锁。
这个方法就是看似复杂,但是其实就是做了一件事,那就是获取该 segment 的独占锁,如果需要的话顺便实例化了一下 node。
3》扩容: rehash
重复一下,segment 数组不能扩容,扩容是 segment 数组某个位置内部的数组 HashEntry\\[] 进行扩容,扩容后,容量为原来的 2 倍。
首先,我们要回顾一下触发扩容的地方,put 的时候,如果判断该值的插入会导致该 segment 的元素个数超过阈值,那么先进行扩容,再插值,读者这个时候可以回去 put 方法看一眼。
该方法不需要考虑并发,因为到这里的时候,是持有该 segment 的独占锁的。
// 方法参数上的 node 是这次扩容后,需要添加到新的数组中的数据。 private void rehash(HashEntry<K,V> node) HashEntry<K,V>[] oldTable = table; int oldCapacity = oldTable.length; // 2 倍 int newCapacity = oldCapacity << 1; threshold = (int)(newCapacity * loadFactor); // 创建新数组 HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 新的掩码,如从 16 扩容到 32,那么 sizeMask 为 31,对应二进制 ‘000...00011111’ int sizeMask = newCapacity - 1; // 遍历原数组,老套路,将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置 for (int i = 0; i < oldCapacity ; i++) // e 是链表的第一个元素 HashEntry<K,V> e = oldTable[i]; if (e != null) HashEntry<K,V> next = e.next; // 计算应该放置在新数组中的位置, // 假设原数组长度为 16,e 在 oldTable[3] 处,那么 idx 只可能是 3 或者是 3 + 16 = 19 int idx = e.hash & sizeMask; if (next == null) // 该位置处只有一个元素,那比较好办 newTable[idx] = e; else // Reuse consecutive sequence at same slot // e 是链表表头 HashEntry<K,V> lastRun = e; // idx 是当前链表的头结点 e 的新位置 int lastIdx = idx; // 下面这个 for 循环会找到一个 lastRun 节点,这个节点之后的所有元素是将要放到一起的 for (HashEntry<K,V> last = next; last != null; last = last.next) int k = last.hash & sizeMask; if (k != lastIdx) lastIdx = k; lastRun = last; // 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置 newTable[lastIdx] = lastRun; // 下面的操作是处理 lastRun 之前的节点, // 这些节点可能分配在另一个链表中,也可能分配到上面的那个链表中 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); // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部 int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable;
,如果没有第一个 for 循环,也是可以工作的,但是,这个 for 循环下来,如果 lastRun 的后面还有比较多的节点,那么这次就是值得的。因为我们只需要克隆 lastRun 前面的节点,后面的一串节点跟着 lastRun 走就是了,不需要做任何操作。
我觉得 Doug Lea 的这个想法也是挺有意思的,不过比较坏的情况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素,那么这次遍历就有点浪费了。不过 Doug Lea 也说了,根据统计,如果使用默认的阈值,大约只有 1/6 的节点需要克隆。
3.2.3、get方法说明
- 计算 hash 值,找到 segment 数组中的具体位置,或我们前面用的“槽”
- 槽中也是一个数组,根据 hash 找到数组中具体的位置
- 到这里是链表了,顺着链表进行查找即可
public V get(Object key) Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; // 1. hash 值 int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 2. 根据 hash 找到对应的 segment if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) // 3. 找到segment 内部数组相应位置的链表,遍历 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;
并发问题
添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的,所以它们之间自然不会有问题,我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。
- put 操作的线程安全性。
- 初始化槽,这个我们之前就说过了,使用了 CAS 来初始化 Segment 中的数组。
- 添加节点到链表的操作是插入到表头的,所以,如果这个时候 get 操作在链表遍历的过程已经到了中间,是不会影响的。当然,另一个并发问题就是 get 操作在 put 之后,需要保证刚刚插入表头的节点被读取,这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。
- 扩容。扩容是新创建了数组,然后进行迁移数据,最后面将 newTable 设置给属性 table。所以,如果 get 操作此时也在进行,那么也没关系,如果 get 先行,那么就是在旧的 table 上做查询操作;而 put 先行,那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。
- remove 操作的线程安全性。
remove 操作我们没有分析源码,所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。
get 操作需要遍历链表,但是 remove 操作会”破坏”链表。
如果 remove 破坏的节点 get 操作已经过去了,那么这里不存在任何问题。
如果 remove 先破坏了一个节点,分两种情况考虑。 1、如果此节点是头结点,那么需要将头结点的 next 设置为数组该位置的元素,table 虽然使用了 volatile 修饰,但是 volatile 并不能提供数组内部操作的可见性保证,所以源码中使用了 UNSAFE 来操作数组,请看方法 setEntryAt。2、如果要删除的节点不是头结点,它会将要删除节点的后继节点接到前驱节点中,这里的并发保证就是 next 属性是 volatile 的。
3.3、jdk8-ConcurrentHashMap
整个 ConcurrentHashMap 由一个个 Segment 组成,Segment 代表”部分“或”一段“的意思,所以很多地方都会将其描述为分段锁用了“槽”来代表一个 segment。
简单理解就是,ConcurrentHashMap 是一个 Segment 数组,Segment 通过继承 ReentrantLock 来进行加锁,所以每次需要加锁的操作锁住的是一个 segment,这样只要保证每个 Segment 是线程安全的,也就实现了全局的线程安全。
Java8 对 ConcurrentHashMap 进行了比较大的改动。建议读者可以参考 Java8 中 HashMap 相对于 Java7 HashMap 的改动,对于 ConcurrentHashMap,Java8 也引入了红黑树。
3.3.1、构造函数
几个构造函数
// 这构造函数里,什么都不干 public ConcurrentHashMap() public ConcurrentHashMap(int initialCapacity) if (initialCapacity < 0) throw new IllegalArgumentException(); int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); this.sizeCtl = cap;
这个初始化方法有点意思,通过提供初始容量,计算了 sizeCtl,sizeCtl = 【 (1.5 * initialCapacity + 1),然后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10,那么得到 sizeCtl 为 16,如果 initialCapacity 为 11,得到 sizeCtl 为 32。
sizeCtl 这个属性使用的场景很多
可以看到原table,现在Node[]的定义
/** * The array of bins. Lazily initialized upon first insertion. * Size is always a power of two. Accessed directly by iterators. */ transient volatile Node<K,V>[] table;
其中Node定义如下
/** * Key-value entry. This class is never exported out as a * user-mutable Map.Entry (i.e., one supporting setValue; see * MapEntry below), but can be used for read-only traversals used * in bulk tasks. Subclasses of Node with a negative hash field * are special, and contain null keys and values (but are never * exported). Otherwise, keys and vals are never null. */ static class Node<K,V> implements Map.Entry<K,V> final int hash; final K key; volatile V val; volatile Node<K,V> next; Node(int hash, K key, V val, Node<K,V> next) this.hash = hash; this.key = key; this.val = val; this.next = next; public final K getKey() return key; public final V getValue() return val; public final int hashCode() return key.hashCode() ^ val.hashCode(); public final String toString() return key + "=" + val; public final V setValue(V value) throw new UnsupportedOperationException(); public final boolean equals(Object o) Object k, v, u; Map.Entry<?,?> e; return ((o instanceof Map.Entry) && (k = (e = (Map.Entry<?,?>)o).getKey()) != null && (v = e.getValue()) != null && (k == key || k.equals(key)) && (v == (u = val) || v.equals(u))); /** * Virtualized support for map.get(); overridden in subclasses. */ 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;
是线程安全的,其通过ReentrantLock 实现
3.3.2、put方法说明
public V put(K key, V value) return putVal(key, value, false); final V putVal(K key, V value, boolean onlyIfAbsent) if (key == null || value == null) throw new NullPointerException(); // 得到 hash 值 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(); // 找该 hash 值对应的数组下标,得到第一个节点 f else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) // 如果数组该位置为空, // 用一次 CAS 操作将这个新值放入其中即可,这个 put 操作差不多就结束了,可以拉到最后面了 // 如果 CAS 失败,那就是有并发操作,进到下一个循环就好了 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin // hash 居然可以等于 MOVED,这个需要到后面才能看明白,不过从名字上也能猜到,肯定是因为在扩容 else if ((fh = f.hash) == MOVED) // 帮助数据迁移,这个等到看完数据迁移部分的介绍后,再理解这个就很简单了 tab = helpTransfer(tab, f); else // 到这里就是说,f 是该位置的头结点,而且不为空 V oldVal = null; // 获取数组该位置的头结点的监视器锁 synchronized (f) if (tabAt(tab, i) == f) if (fh >= 0) // 头结点的 hash 值大于 0,说明是链表 // 用于累加,记录链表的长度 binCount = 1; // 遍历链表 for (Node<K,V> e = f;; ++binCount) K ek; // 如果发现了"相等"的 key,判断是否要进行值覆盖,然后也就可以 break 了 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; // binCount != 0 说明上面在做链表操作 if (binCount != 0) // 判断是否要将链表转换为红黑树,临界值和 HashMap 一样,也是 8 if (binCount >= TREEIFY_THRESHOLD) // 这个方法和 HashMap 中稍微有一点点不同,那就是它不是一定会进行红黑树转换, // 如果当前数组的长度小于 64,那么会选择进行数组扩容,而不是转换为红黑树 // 具体源码我们就不看了,扩容部分后面说 treeifyBin(tab, i); if (oldVal != null) return oldVal; break; // addCount(1L, binCount); return null;
第一个是初始化,第二个是扩容,第三个是帮助数据迁移
1》初始化数组:initTable
这个比较简单,主要就是初始化一个合适大小的数组,然后会设置 sizeCtl。
初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。
private final Node<K,V>[] initTable() Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) // 初始化的"功劳"被其他线程"抢去"了 if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin // CAS 一下,将 sizeCtl 设置为 -1,代表抢到了锁 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) try if ((tab = table) == null || tab.length == 0) // DEFAULT_CAPACITY 默认初始容量是 16 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 初始化数组,长度为 16 或初始化时提供的长度 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 将这个数组赋值给 table,table 是 volatile 的 table = tab = nt; // 如果 n 为 16 的话,那么这里 sc = 12 // 其实就是 0.75 * n sc = n - (n >>> 2); finally // 设置 sizeCtl 为 sc,我们就当是 12 吧 sizeCtl = sc; break; return tab;
2》链表转红黑树: treeifyBin
前面我们在 put 源码分析也说过,treeifyBin 不一定就会进行红黑树转换,也可能是仅仅做数组扩容。
private final void treeifyBin(Node<K,V>[] tab, int index) Node<K,V> b; int n, sc; if (tab != null) // MIN_TREEIFY_CAPACITY 为 64 // 所以,如果数组长度小于 64 的时候,其实也就是 32 或者 16 或者更小的时候,会进行数组扩容 if ((n = tab.length) < MIN_TREEIFY_CAPACITY) // 后面我们再详细分析这个方法 tryPresize(n << 1); // b 是头结点 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));
3》扩容:tryPresize
如果说 Java8 ConcurrentHashMap 的源码不简单,那么说的就是扩容操作和迁移操作。
这个方法要完完全全看懂还需要看之后的 transfer 方法,读者应该提前知道这点。
这里的扩容也是做翻倍扩容的,扩容后数组容量为原来的 2 倍。
// 首先要说明的是,方法参数 size 传进来的时候就已经翻了倍了 private final void tryPresize(int size) // c:size 的 1.5 倍,再加 1,再往上取最近的 2 的 n 次方。 int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1); int sc; while ((sc = sizeCtl) >= 0) Node<K,V>[] tab = table; int n; // 这个 if 分支和之前说的初始化数组的代码基本上是一样的,在这里,我们可以不用管这块代码 if (tab == null || (n = tab.length) == 0) n = (sc > c) ? sc : c; if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) try if (table == tab) @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = nt; sc = n - (n >>> 2); // 0.75 * n finally sizeCtl = sc; else if (c <= sc || n >= MAXIMUM_CAPACITY) break; else if (tab == table) // 我没看懂 rs 的真正含义是什么,不过也关系不大 int rs = resizeStamp(n); if (sc < 0) Node<K,V>[] nt; if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; // 2. 用 CAS 将 sizeCtl 加 1,然后执行 transfer 方法 // 此时 nextTab 不为 null if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); // 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2) // 我是没看懂这个值真正的意义是什么?不过可以计算出来的是,结果是一个比较大的负数 // 调用 transfer 方法,此时 nextTab 参数为 null else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null);
这个方法的核心在于 sizeCtl 值的操作,首先将其设置为一个负数,然后执行 transfer(tab, null),再下一个循环将 sizeCtl 加 1,并执行 transfer(tab, nt),之后可能是继续 sizeCtl 加 1,并执行 transfer(tab, nt)。
所以,可能的操作就是执行 1 次 transfer(tab, null) + 多次 transfer(tab, nt),这里怎么结束循环的需要看完 transfer 源码才清楚。
4》
数据迁移:transfer
下面这个方法很点长,将原来的 tab 数组的元素迁移到新的 nextTab 数组中。
虽然我们之前说的 tryPresize 方法中多次调用 transfer 不涉及多线程,但是这个 transfer 方法可以在其他地方被调用,典型地,我们之前在说 put 方法的时候就说过了,请往上看 put 方法,是不是有个地方调用了 helpTransfer 方法,helpTransfer 方法会调用 transfer 方法的。
此方法支持多线程执行,外围调用此方法的时候,会保证第一个发起数据迁移的线程,nextTab 参数为 null,之后再调用此方法的时候,nextTab 不会为 null。
阅读源码之前,先要理解并发操作的机制。原数组长度为 n,所以我们有 n 个迁移任务,让每个线程每次负责一个小任务是最简单的,每做完一个任务再检测是否有其他没做完的任务,帮助迁移就可以了,而 Doug Lea 使用了一个 stride,简单理解就是步长,每个线程每次负责迁移其中的一部分,如每次迁移 16 个小任务。所以,我们就需要一个全局的调度者来安排哪个线程执行哪几个任务,这个就是属性 transferIndex 的作用。
第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置,然后从后往前的 stride 个任务属于第一个线程,然后将 transferIndex 指向新的位置,再往前的 stride 个任务属于第二个线程,依此类推。当然,这里说的第二个线程不是真的一定指代了第二个线程,也可以是同一个线程,这个读者应该能理解吧。其实就是将一个大的迁移任务分为了一个个任务包。
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) int n = tab.length, stride; // stride 在单核下直接等于 n,多核模式下为 (n>>>3)/NCPU,最小值是 16 // stride 可以理解为”步长“,有 n 个位置是需要进行迁移的, // 将这 n 个任务分为多个任务包,每个任务包有 stride 个任务 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range // 如果 nextTab 为 null,先进行一次初始化 // 前面我们说了,外围会保证第一个发起迁移的线程调用此方法时,参数 nextTab 为 null // 之后参与迁移的线程调用此方法时,nextTab 不会为 null if (nextTab == null) try // 容量翻倍 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 是 ConcurrentHashMap 中的属性 nextTable = nextTab; // transferIndex 也是 ConcurrentHashMap 的属性,用于控制迁移的位置 transferIndex = n; int nextn = nextTab.length; // ForwardingNode 翻译过来就是正在被迁移的 Node // 这个构造方法会生成一个Node,key、value 和 next 都为 null,关键是 hash 为 MOVED // 后面我们会看到,原数组中位置 i 处的节点完成迁移工作后, // 就会将位置 i 处设置为这个 ForwardingNode,用来告诉其他线程该位置已经处理过了 // 所以它其实相当于是一个标志。 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // advance 指的是做完了一个位置的迁移工作,可以准备做下一个位置的了 boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab /* * 下面这个 for 循环,最难理解的在前面,而要看懂它们,应该先看懂后面的,然后再倒回来看 * */ // i 是位置索引,bound 是边界,注意是从后往前 for (int i = 0, bound = 0;;) Node<K,V> f; int fh; // 下面这个 while 真的是不好理解 // advance 为 true 表示可以进行下一个位置的迁移了 // 简单理解结局:i 指向了 transferIndex,bound 指向了 transferIndex-stride while (advance) int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; // 将 transferIndex 值赋给 nextIndex // 这里 transferIndex 一旦小于等于 0,说明原数组的所有位置都有相应的线程去处理了 else if ((nextIndex = transferIndex) <= 0) i = -1; advance = false; else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) // 看括号中的代码,nextBound 是这次迁移任务的边界,注意,是从后往前 bound = nextBound; i = nextIndex - 1; advance = false; if (i < 0 || i >= n || i + n >= nextn) int sc; if (finishing) // 所有的迁移操作已经完成 nextTable = null; // 将新的 nextTab 赋值给 table 属性,完成迁移 table = nextTab; // 重新计算 sizeCtl:n 是原数组长度,所以 sizeCtl 得出的值将是新数组长度的 0.75 倍 sizeCtl = (n << 1) - (n >>> 1); return; // 之前我们说过,sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2 // 然后,每有一个线程参与迁移就会将 sizeCtl 加 1, // 这里使用 CAS 操作对 sizeCtl 进行减 1,代表做完了属于自己的任务 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) // 任务结束,方法退出 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 到这里,说明 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT, // 也就是说,所有的迁移任务都做完了,也就会进入到上面的 if(finishing) 分支了 finishing = advance = true; i = n; // recheck before commit // 如果位置 i 处是空的,没有任何节点,那么放入刚刚初始化的 ForwardingNode ”空节点“ else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); // 该位置处是一个 ForwardingNode,代表该位置已经迁移过了 else if ((fh = f.hash) == MOVED) advance = true; // already processed else // 对数组该位置处的结点加锁,开始处理数组该位置处的迁移工作 synchronized (f) if (tabAt(tab, i) == f) Node<K,V> ln, hn; // 头结点的 hash 大于 0,说明是链表的 Node 节点 if (fh >= 0) // 下面这一块和 Java7 中的 ConcurrentHashMap 迁移是差不多的, // 需要将链表一分为二, // 找到原链表中的 lastRun,然后 lastRun 及其之后的节点是一起进行迁移的 // lastRun 之前的节点需要进行克隆,然后分到两个链表中 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); // 其中的一个链表放在新数组的位置 i setTabAt(nextTab, i, ln); // 另一个链表放在新数组的位置 i+n setTabAt(nextTab, i + n, hn); // 将原数组该位置处设置为 fwd,代表该位置已经处理完毕, // 其他线程一旦看到该位置的 hash 值为 MOVED,就不会进行迁移了 setTabAt(tab, i, fwd); // advance 设置为 true,代表该位置已经迁移完毕 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; // 如果一分为二后,节点数少于 8,那么将红黑树转换回链表 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; // 将 ln 放置在新数组的位置 i setTabAt(nextTab, i, ln); // 将 hn 放置在新数组的位置 i+n setTabAt(nextTab, i + n, hn); // 将原数组该位置处设置为 fwd,代表该位置已经处理完毕, // 其他线程一旦看到该位置的 hash 值为 MOVED,就不会进行迁移了 setTabAt(tab, i, fwd); // advance 设置为 true,代表该位置已经迁移完毕 advance = true;
transfer 这个方法并没有实现所有的迁移任务,每次调用这个方法只实现了 transferIndex 往前 stride 个位置的迁移工作,其他的需要由外围来控制。
3.2.3、get方法说明
- 计算 hash 值
- 根据 hash 值找到数组对应位置: (n – 1) & h
- 根据该位置处结点性质进行相应查找
- 如果该位置为 null,那么直接返回 null 就可以了
- 如果该位置处的节点刚好就是我们需要的,返回该节点的值即可
- 如果该位置节点的 hash 值小于 0,说明正在扩容,或者是红黑树,后面我们再介绍 find 方法
- 如果以上 3 条都不满足,那就是链表,进行遍历比对即可
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; // 如果头结点的 hash 小于 0,说明 正在扩容,或者该位置是红黑树 else if (eh < 0) // 参考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k) 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;
参看地址:
http://www.importnew.com/28263.html
go语言学习之旅--并发编程
Go语言学习之旅--并发编程golang并发编程之协程golang并发编程之通道golang并发编程之WaitGroup实现同步golang并发编程之runtime包golang并发编程之Mutex互斥锁实现同步golang并发编程之channel的遍历golang并发编程之selectswitchgolang并发编程之T... 查看详情
go语言学习之旅--并发编程
Go语言学习之旅--并发编程golang并发编程之协程golang并发编程之通道golang并发编程之WaitGroup实现同步golang并发编程之runtime包golang并发编程之Mutex互斥锁实现同步golang并发编程之channel的遍历golang并发编程之selectswitchgolang并发编程之T... 查看详情
并发编程路线
一python并发编程之多进程理论二python并发编程之多进程实现三python并发编程之多线程理论四python并发编程之多线程实现五python并发编程之协程理论六python并发编程之协程实现七python并发编程之IO模型五补充:paramiko模块 查看详情
go语言学习之旅--并发编程
Go语言学习之旅--并发编程golang并发编程之协程golang并发编程之通道golang并发编程之WaitGroup实现同步golang并发编程之runtime包golang并发编程之Mutex互斥锁实现同步golang并发编程之channel的遍历golang并发编程之selectswitchgolang并发编程之T... 查看详情
java并发编程之美
一、内容简介并发编程相比Java中其他知识点的学习门槛较高,从而导致很多人望而却步。但无论是职场面试,还是高并发/高流量系统的实现,却都离不开并发编程,于是能够真正掌握并发编程的人成为了市场迫切需求的人才。... 查看详情
java并发编程:核心理论
Java并发编程系列:Java并发编程:核心理论 Java并发编程:Synchronized及其实现原理Java并发编程:Synchronized底层优化(轻量级锁、偏向锁)Java并发编程:线程间的协作(wait/notify/sleep/yield/join)Java并发编程:volatile的使用及其原理... 查看详情
并发编程的优缺点
为什么要使用并发编程(并发编程的优点)充分利用多核CPU的计算能力方便进行业务拆分,提升系统并发能力和性能并发编程的缺点并发编程的目的就是为了能提高程序的执行效率,提高程序运行速度,但是并发编程并不总是能... 查看详情
java并发编程:并发编程的认识(代码片段)
1、并发编程简介并发编程:所谓并发编程是指在一台处理器上“同时”处理多个任务。并发是在同一实体上的多个事件。多个事件在同一时间间隔发生。并发编程的目标:充分的利用处理器的每一个核,以达到最高... 查看详情
ribbon超时设置
...时时间,在图上调用的时候肯定是报超时错误:java.util.concurrent.TimeoutException:null。三、设置超时1.在shop-gateway的application.properties中添加配置,设置的ReadTimeout是8秒,网上看的,说是全局设置:ribbon.ReadTimeout=8000ribbon.ConnectTimeout=600... 查看详情
初识并发编程
...家都会使用Executors工具类创建线程池来使用线程,为什么并发编程会使得程序更快?那它的难点又在哪里?本文首先介绍何为并发编程。一、并发编程 并发编程的目的是为了让程序运行得更快,但是,并不是启动更多的线程... 查看详情
并发编程:我对java并发编程的总结和思考
编写优质的并发代码是一件难度极高的事情。Java语言从第一版本开始内置了对多线程的支持,这一点在当年是非常了不起的,但是当我们对并发编程有了更深刻的认识和更多的实践后,实现并发编程就有了更多的方案和更好的... 查看详情
高并发编程必备基础--转载自并发编程网
文章转载自并发编程网 本文链接地址:高并发编程必备基础一、前言借用Java并发编程实践中的话”编写正确的程序并不容易,而编写正常的并发程序就更难了”,相比于顺序执行的情况,多线程的线程安全问题是微妙而且... 查看详情
网络编程和并发编程
1.网络编程小白入门知识点大杂烩 2.并发编程小白入门知识点大杂烩 查看详情
java并发编程
Java并发编程(一)之前看《ThinkingInJava》时,并发讲解的挺多的,自己算是初步了解了并发。但是其讲解的不深入,自己感觉其讲解的不够好。后来自己想再学一学并发,买了《Java并发编程实战》,看了一下讲的好基础、好多的理... 查看详情
并发编程
同步异步进程与线程多线程并发编程高阶IO模型 查看详情
解决并发编程之痛的良药--结构化并发编程(代码片段)
解决并发编程之痛的良药--结构化并发编程作者简介:曹家锋,Westar实验室技术专家。Westar实验室(westar.io),成立于2018年,关注于区块链及分布式前沿技术,包括区块链分层架构、二层路由,网络性能、智能合约、PoW优化等。... 查看详情
java并发编程:synchronized及其实现原理
Java并发编程系列:Java并发编程:核心理论 Java并发编程:Synchronized及其实现原理Java并发编程:Synchronized底层优化(轻量级锁、偏向锁)Java并发编程:线程间的协作(wait/notify/sleep/yield/join)Java并发编程:volatile的使用及其原理... 查看详情
并发编程基础
并发编程技术:多线程,异步编程,并行处理,TPL数据流,响应式编程。 并发:同时做多件事。多线程:并发的一种形式,采用多个线程来执行程序。异步编程:并发的一种形式,采用future模式或回调callback机制,以避免产... 查看详情