java

#java| 来源: 网络整理| 查看: 265

java-浅学HashTable与ConcurrentHashMap

目录 java-浅学HashTable与ConcurrentHashMapHashtableConcurrentHashMapJDK1.7与JDK1.8区别ConcurrentHashMap1.7put（K，V）rehash（）扩容get（K） ConcurrentHashMap1.8put（K，V）get（K）

Hashtable

HashTable也是一种key-value结构，它继承自Dictionary，实现了Map和Cloneable以及Serializable接口。

其继承体系结构图如下图所示。

在结构上Hashtable一直是由数组+链表形成，与HashMap一样，Hashtable在哈希表中存储键值对，但是Hashtable不允许键或值为null。

Hashtable和hashMap类很相似，但是Hashtable支持同步，它在几乎所有的方法上都加上了synchronized锁，而加锁的结果就是HashTable操作的效率十分低下。

要避免 HashMap 的线程安全问题，有多个解决方法，比如改用 HashTable 、 Collections.synchronizedMap（）方法以及ConcurrentHashMap（）方法。

ConcurrentHashMap

ConcurrentHashMap也被称为线程安全的HashMap，在生产开发中使用的频率也是较高的，与HashMap一样，JDK1.7前与JDK1.8之后的ConcurrentHashMap会有一些不同，接下来让我们浅浅了解一下吧

JDK1.7与JDK1.8区别

JDK1.7之前，ConcurrentHashMap使用Segment+数组+链表的结构，每个Segment对应一把锁，如果多个线程访问不同的Segment，则不会冲突。

一个Segment其实就是一个类似HashMap 的结构，Segment内部维护了一个链表数组，我们用下面这一幅图来看下ConcurrentHashMap的内部结构：

ConcurrnetHashMap由很多个Segment组合，而每一个Segment是一个类似于 HashMap 的结构，所以每一个 HashMap的内部可以进行扩容。但是Segment的个数一旦初始化就不能改变，默认 Segment的个数是 16 个，你也可以认为ConcurrentHashMap默认支持最多 16 个线程并发。

JDK1.8开始，ConcurrentHashMap将数组的每个头节点作为锁，如果多个线程访问的头节点不同，则不会冲突。并且在结构上，和HashMap一样，从数组+链表改为数组+链表+红黑树。

和 JDK1.7 的 HasEntry 作用相同，对 val 和 next 都用了 volatile 关键字，保证了可见性。

ConcurrentHashMap1.7 put（K，V）计算要 put 的 key 的位置，获取指定位置的 Segment。如果指定位置的 Segment 为空，则初始化这个 Segment。

初始化 Segment 流程：

检查计算得到的位置的 Segment 是否为null.为 null 继续初始化，使用 Segment[0] 的容量和负载因子创建一个 HashEntry 数组。再次检查计算得到的指定位置的 Segment 是否为null.使用创建的 HashEntry 数组初始化这个 Segment.自旋判断计算得到的指定位置的 Segment 是否为null，使用 CAS 在这个位置赋值为 Segment. /** * Maps the specified key to the specified value in this table. * Neither the key nor the value can be null. * *

The value can be retrieved by calling the get method * with a key that is equal to the original key. * * @param key key with which the specified value is to be associated * @param value value to be associated with the specified key * @return the previous value associated with key, or * null if there was no mapping for key * @throws NullPointerException if the specified key or value is null */ public V put(K key, V value) { Segment s; if (value == null) throw new NullPointerException(); int hash = hash(key); // hash 值无符号右移 28位（初始化时获得），然后与 segmentMask=15 做与运算 // 其实也就是把高4位与segmentMask（1111）做与运算 int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment)UNSAFE.getObject // nonvolatile; recheck (segments, (j Segment proto = ss[0]; // use segment 0 as prototype // 获取0号 segment 里的 HashEntry 初始化长度 int cap = proto.table.length; // 获取0号 segment 里的 hash 表里的扩容负载因子，所有的 segment 的 loadFactor 是相同的 float lf = proto.loadFactor; // 计算扩容阀值 int threshold = (int)(cap * lf); // 创建一个 cap 容量的 HashEntry 数组 HashEntry[] tab = (HashEntry[])new HashEntry[cap]; if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck // 再次检查 u 位置的 Segment 是否为null，因为这时可能有其他线程进行了操作 Segment s = new Segment(lf, threshold, tab); // 自旋检查 u 位置的 Segment 是否为null while ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) { // 使用CAS 赋值，只会成功一次 if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }

上面获取Segment和初始化Segment的操作，接下来将查看Segment的put方法。

final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 获取 ReentrantLock 独占锁，获取不到，scanAndLockForPut 获取。 HashEntry node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry[] tab = table; // 计算要put的数据位置 int index = (tab.length - 1) & hash; // CAS 获取 index 坐标的值 HashEntry first = entryAt(tab, index); for (HashEntry e = first;;) { if (e != null) { // 检查是否 key 已经存在，如果存在，则遍历链表寻找位置，找到后替换 value 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 { // first 有值没说明 index 位置已经有值了，有冲突，链表头插法。 if (node != null) node.setNext(first); else node = new HashEntry(hash, key, value, first); int c = count + 1; // 容量大于扩容阀值，小于最大容量，进行扩容 if (c > threshold && tab.length HashEntry[] oldTable = table; // 老容量 int oldCapacity = oldTable.length; // 新容量，扩大两倍 int newCapacity = oldCapacity HashEntry 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 lastRun = e; int lastIdx = idx; // 新的位置只可能是不便或者是老的位置+老的容量。 // 遍历结束后，lastRun 后面的元素位置都是相同的 for (HashEntry last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } // ，lastRun 后面的元素位置都是相同的，直接作为链表赋值到新位置。 newTable[lastIdx] = lastRun; // Clone remaining nodes for (HashEntry p = e; p != lastRun; p = p.next) { // 遍历剩余元素，头插法到指定 k 位置。 V v = p.value; int h = p.hash; int k = h & sizeMask; HashEntry n = newTable[k]; newTable[k] = new HashEntry(h, p.key, v, n); } } } } // 头插法插入新的节点 int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; } get（K）计算得到 key 的存放位置。遍历指定位置查找相同 key 的 value 值。 public V get(Object key) { Segment s; // manually integrate access methods to reduce overhead HashEntry[] tab; int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) // 如果是链表，遍历查找到相同 key 的 value。 K k; if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; } ConcurrentHashMap1.8 put（K，V）

初始化initTable：

/** * Initializes table, using the size recorded in sizeCtl. */ private final Node[] initTable() { Node[] tab; int sc; while ((tab = table) == null || tab.length == 0) { //如果 sizeCtl < 0 ,说明另外的线程执行CAS 成功，正在进行初始化。 if ((sc = sizeCtl) if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") Node[] nt = (Node[])new Node[n]; table = tab = nt; sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; }

从源码中可以发现 ConcurrentHashMap 的初始化是通过自旋和 CAS 操作完成的。里面需要注意的是变量 sizeCtl ，它的值决定着当前的初始化状态。sizeCtl采用volatile进行修饰，确保线程间可见性。

-1 说明正在初始化-N 说明有N-1个线程正在进行扩容表示 table 初始化大小，如果 table 没有初始化表示 table 容量，如果 table　已经初始化

接下来就是put方法了，分三种情况讨论

根据 key 计算出 hashcode 。判断是否需要进行初始化。即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。如果都不满足，则利用 synchronized 锁写入数据。如果数量大于 TREEIFY_THRESHOLD 则要执行树化方法，在treeifyBin中会首先判断当前数组长度≥64时才会将链表转换为红黑树 public V put(K key, V value) { return putVal(key, value, false); } /** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { // key 和 value 不能为空 if (key == null || value == null) throw new NullPointerException(); int hash = spread(key.hashCode()); int binCount = 0; for (Node[] tab = table;;) { // f = 目标位置元素 Node f; int n, i, fh;// fh 后面存放目标位置的元素 hash 值 if (tab == null || (n = tab.length) == 0) // 数组桶为空，初始化数组桶（自旋+CAS) tab = initTable(); //一种特殊的节点（forwarding 节点，迁移节点，只在迁移过程中存在）的处理方式 //ForwardingNode：用于解决当进行扩容的时候，进行查询的问题。标记数组中的链表已经被处理过了，就可以不用重复处理 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 桶内为空，CAS 放入，不加锁，成功了就直接 break 跳出 if (casTabAt(tab, i, null,new Node(hash, key, value, null))) break; // no lock when adding to empty bin } else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; // 使用 synchronized 加锁加入节点 synchronized (f) { if (tabAt(tab, i) == f) { // 说明是链表 if (fh >= 0) { binCount = 1; // 循环加入新的或者覆盖节点 for (Node 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 pred = e; if ((e = e.next) == null) { pred.next = new Node(hash, key, value, null); break; } } } else if (f instanceof TreeBin) { // 红黑树 Node p; binCount = 2; if ((p = ((TreeBin)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null; } get（K）根据 hash 值计算位置。查找到指定位置，如果头节点就是要找的，直接返回它的 value.如果头节点 hash 值小于 0 ，说明正在扩容或者是红黑树，查找之。如果是链表，遍历查找之。 /* * 源码中没有一处加锁，却能保证线程安全，Node的成员val是用volatile修饰，保证了线程间的可见性。 * 成员val是用volatile修饰和数组用volatile修饰没有关系，数组用volatile修饰主要是保证在数组扩容的时候保证可见性。 */ public V get(Object key) { //大佬们写代码都是在使用的时候才初始化，非常优雅 Node[] tab; Node e, p; int n, eh; K ek; //计算 key 所在的 hash 位置 int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { // 如果指定位置元素存在，头结点hash值相同 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) // key hash 值相等，key值相同，直接返回元素 value return e.val; } //hash值为负值表示正在扩容，这个时候查的是ForwardingNode的find方法来定位到nextTable来 //eh=-1，说明该节点是一个ForwardingNode，正在迁移，此时调用ForwardingNode的find方法去nextTable里找。 //eh=-2，说明该节点是一个TreeBin，此时调用TreeBin的find方法遍历红黑树，由于红黑树有可能正在旋转变色，所以find里会有读写锁。 //eh>=0，说明该节点下挂的是一个链表，直接遍历该链表即可。 else if (eh

【本文地址】

java

java

今日新闻

推荐新闻