Concurrenthashmap

简介ConcurrentHashMap是HashMap的线程安全版本，内部也是使用（数组 + 链表 + 红黑树）的结构来存储元素。 相比于同样线程安全的HashTable来说，效率等各方面都有极大地提高。 用到锁的简介这里先简单介绍一下各种锁，以便下文讲到相关概念时能有个印象。 synchronized java中的关键字，内部实现为监视器锁，主要是通过对象监视器在对象头中的字段来表明的。 synchronized从旧版本到现在已经做了很多优化了，在运行时会有三种存在方式：偏向锁，轻量级锁，重量级锁。 偏向锁，是指一段同步代码一直被一个线程访问，那么这个线程会自动获取锁，降低获取锁的代价。 轻量级锁，是指当锁是偏向锁时，被另一个线程所访问，偏向锁会升级为轻量级锁，这个线程会通过自旋的方式尝试获取锁，不会阻塞，提高性能。 重量级锁，是指当锁是轻量级锁时，当自旋的线程自旋了一定的次数后，还没有获取到锁，就会进入阻塞状态，该锁升级为重量级锁，重量级锁会使其他线程阻塞，性能降低。CAS CAS，Compare And Swap，它是一种乐观锁，认为对于同一个数据的并发操作不一定会发生修改，在更新数据的时候，尝试去更新数据，如果失败就不断尝试。volatile（非锁） java中的关键字，当多个线程访问同一个变量时，一个线程修改了这个变量的值，其他线程能够立即看得到修改的值。 volatile只保证可见性，不保证原子性，比如 volatile修改的变量 i，针对i++操作，不保证每次结果都正确，因为i++操作是两步操作，相当于 i = i +1，先读取，再加1，这种情况 volatile是无法保证的。自旋锁 自旋锁，是指尝试获取锁的线程不会阻塞，而是循环的方式不断尝试，这样的好处是减少线程的上下文切换带来的开锁，提高性能，缺点是循环会消耗CPU。分段锁 分段锁，是一种锁的设计思路，它细化了锁的粒度，主要运用在ConcurrentHashMap中，实现高效的并发操作，当操作不需要更新整个数组时，就只锁数组中的一项就可以了。ReentrantLock 可重入锁，是指一个线程获取锁之后再尝试获取锁时会自动获取锁，可重入锁的优点是避免死锁，synchronized也是可重入锁。源码分析构造方法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;}public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this.sizeCtl = DEFAULT_CAPACITY; putAll(m);}public ConcurrentHashMap(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, 1);}public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap;}构造方法与HashMap对比可以发现，没有了HashMap中的threshold和loadFactor，而是改用了sizeCtl来控制，而且只存储了容量在里面，官方给出的解释如下： ...

删除元素删除元素跟添加元素一样，都是先找到元素所在的桶，然后采用分段锁的思想锁住整个桶，再进行操作。 public V remove(Object key) { // 调用替换节点方法 return replaceNode(key, null, null);}final V replaceNode(Object key, V value, Object cv) { // 计算hash int hash = spread(key.hashCode()); // 自旋 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0 || (f = tabAt(tab, i = (n - 1) & hash)) == null) // 如果目标key所在的桶不存在，跳出循环返回null break; else if ((fh = f.hash) == MOVED) // 如果正在扩容中，协助扩容 tab = helpTransfer(tab, f); else { V oldVal = null; // 标记是否处理过 boolean validated = false; synchronized (f) { // 再次验证当前桶第一个元素是否被修改过 if (tabAt(tab, i) == f) { if (fh >= 0) { // fh>=0表示是链表节点 validated = true; // 遍历链表寻找目标节点 for (Node<K,V> e = f, pred = null;;) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { // 找到了目标节点 V ev = e.val; // 检查目标节点旧value是否等于cv if (cv == null || cv == ev || (ev != null && cv.equals(ev))) { oldVal = ev; if (value != null) // 如果value不为空则替换旧值 e.val = value; else if (pred != null) // 如果前置节点不为空 // 删除当前节点 pred.next = e.next; else // 如果前置节点为空 // 说明是桶中第一个元素，删除之 setTabAt(tab, i, e.next); } break; } pred = e; // 遍历到链表尾部还没找到元素，跳出循环 if ((e = e.next) == null) break; } } else if (f instanceof TreeBin) { // 如果是树节点 validated = true; TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> r, p; // 遍历树找到了目标节点 if ((r = t.root) != null && (p = r.findTreeNode(hash, key, null)) != null) { V pv = p.val; // 检查目标节点旧value是否等于cv if (cv == null || cv == pv || (pv != null && cv.equals(pv))) { oldVal = pv; if (value != null) // 如果value不为空则替换旧值 p.val = value; else if (t.removeTreeNode(p)) // 如果value为空则删除元素 // 如果删除后树的元素个数较少则退化成链表 // t.removeTreeNode(p)这个方法返回true表示删除节点后树的元素个数较少 setTabAt(tab, i, untreeify(t.first)); } } } } } // 如果处理过，不管有没有找到元素都返回 if (validated) { // 如果找到了元素，返回其旧值 if (oldVal != null) { // 如果要替换的值为空，元素个数减1 if (value == null) addCount(-1L, -1); return oldVal; } break; } } } // 没找到元素返回空 return null;}计算hash；如果所在的桶不存在，表示没有找到目标元素，返回；如果正在扩容，则协助扩容完成后再进行删除操作；如果是以链表形式存储的，则遍历整个链表查找元素，找到之后再删除；如果是以树形式存储的，则遍历树查找元素，找到之后再删除；如果是以树形式存储的，删除元素之后树较小，则退化成链表；如果确实删除了元素，则整个map元素个数减1，并返回旧值；如果没有删除元素，则返回null；获取元素获取元素，根据目标key所在桶的第一个元素的不同采用不同的方式获取元素，关键点在于find()方法的重写。 ...

上一篇文章已经就ConcurrentHashMap进行了部分说明，介绍了其中涉及的常量和变量的含义，有些部分需要结合方法源码来理解，今天这篇文章就继续讲解并发ConcurrentHashMap 前言本文主要介绍ConcurrentHashMap中的一些重要方法，结合上篇文章中的讲解部分进行更进一步的介绍 回顾下上篇文章，我们应该已经知道ConcurrentHashMap的整体结构和HashMap基本一致，不同的是处理多线程并发下保证操作的正确性，ConcurrentHashMap通过CAS和synchronized进行并发控制，当然，这种情况下各种处理都会变的更为复杂，下面我们就通过方法来深入理解ConcurrentHashMap的操作 重要方法在一些方法中展示了各个变量以及常量的使用，能让我们更好的理解其中的操作 tabAt/casTabAt/setTabAt下列方法用于读写table数组，使用Unsafe提供的更新获取volatile变量，CAS更新数组元素等操作 // 读取table[i] @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); } // CAS更新table[i] 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); } // 插入table[i] static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); }sizesize方法返回了一个不精确的值，在多线程环境下，返回一个不精确的值，通过sumCount迭代counterCells统计sum值。 ...

ConcurrentHashMapConcurrentHashMap是线程安全，性能出色的Map的线程安全实现，相比较HashMap他是线程安全的，相比较HashTable他的性能优势非常明显。他的使用很简单，这里主要是想要探究一下ConcurrentHashMap的实现原理。在这里一共有 个问题需要搞明白。 ConcurrentHashMap为什么比HashTable的性能要高？ConcurrentHashMap在JDK8和JDK7有什么变化，为什么会有这种变化，对我们开发有什么启示？为什么在JDK8中使用Synchronized而不是用ReentrantLock来实现加锁？带着这几个问题我们来分析一下ConcurrentHashMap的源码吧。 ConcurrentHashMap定义在JDK8中ConcurrentHashMap的定义如下： public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> implements ConcurrentMap<K,V>, Serializable {

在学习ConcurrentHashMap时发现，源码中对table数组的元素进行操作时，使用了三个封装好的原子操作方法，如下： /* ---------------- Table element access -------------- *//* * Atomic access methods are used for table elements as well as * elements of in-progress next table while resizing. All uses of * the tab arguments must be null checked by callers. All callers * also paranoically precheck that tab's length is not zero (or an * equivalent check), thus ensuring that any index argument taking * the form of a hash value anded with (length - 1) is a valid * index. Note that, to be correct wrt arbitrary concurrency * errors by users, these checks must operate on local variables, * which accounts for some odd-looking inline assignments below. * Note that calls to setTabAt always occur within locked regions, * and so require only release ordering. */@SuppressWarnings("unchecked")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);}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);}static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectRelease(tab, ((long)i << ASHIFT) + ABASE, v);}casTabAt这个方法我们可以很清晰地明白它封装了对于数组元素的cas操作，但是另外两个方法的意义如何理解呢？ ...