在上一篇中介绍了hashmap的原理，这一节是concurrentmap的最后一节，所以会完整的介绍concurrenthashmap的实现。
concurrenthashmap原理
在读写锁章节部分介绍过一种是用读写锁实现map的方法。此种方法看起来可以实现map响应的功能，而且吞吐量也应该不错。但是通过前面对读写锁原理的分析后知道，读写锁的适合场景是读操作>>写操作，也就是读操作应该占据大部分操作，另外读写锁存在一个很严重的问题是读写操作不能同时发生。要想解决读写同时进行问题（至少不同元素的读写分离），那么就只能将锁拆分，不同的元素拥有不同的锁，这种技术就是“锁分离”技术。
默认情况下concurrenthashmap是用了16个类似hashmap 的结构，其中每一个hashmap拥有一个独占锁。也就是说最终的效果就是通过某种hash算法，将任何一个元素均匀的映射到某个hashmap的map.entry上面，而对某个一个元素的操作就集中在其分布的hashmap上，与其它hashmap无关。这样就支持最多16个并发的写操作。
上图就是concurrenthashmap的类图。参考上面的说明和hashmap的原理分析，可以看到concurrenthashmap将整个对象列表分为segmentmask 1个片段（segment）。其中每一个片段是一个类似于hashmap的结构，它有一个hashentry的数组，数组的每一项又是一个链表，通过hashentry的next引用串联起来。
这个类图上面的数据结构的定义非常有学问，接下来会一个个有针对性的分析。
首先如何从concurrenthashmap定位到hashentry。在hashmap的原理分析部分说过，对于一个hash的数据结构来说，为了减少浪费的空间和快速定位数据，那么就需要数据在hash上的分布比较均匀。对于一次map的查找来说，首先就需要定位到segment，然后从过segment定位到hashentry链表，最后才是通过遍历链表得到需要的元素。
在不讨论并发的前提下先来讨论如何定位到hashentry的。在concurrenthashmap中是通过hash(key.hashcode())和segmentfor(hash)来得到segment的。清单1 描述了如何定位segment的过程。其中hash(int)是将key的hashcode进行二次编码，使之能够在segmentmask 1个segment上均匀分布（默认是16个）。可以看到的是这里和hashmap还是有点不同的，这里采用的算法叫wang/jenkins hash，有兴趣的可以和。总之它的目的就是使元素能够均匀的分布在不同的segment上，这样才能够支持最多segmentmask 1个并发，这里segmentmask 1是segments的大小。
清单1 定位segment
private static int hash(int h) {
// spread bits to regularize both segment and index locations,
// using variant of single-word wang/jenkins hash.
h = (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h = (h << 3);
h ^= (h >>> 6);
h = (h << 2) (h << 14);
return h ^ (h >>> 16);
}
final segment segmentfor(int hash) {
return segments[(hash >>> segmentshift) & segmentmask];
}
显然在不能够对segment扩容的情况下，segments的大小就应该是固定的。所以在concurrenthashmap中segments/segmentmask/segmentshift都是常量，一旦初始化后就不能被再次修改，其中segmentshift是查找segment的一个常量偏移量。
有了segment以后再定位hashentry就和hashmap中定位hashentry一样了，先将hash值与segment中hashentry的大小减1进行与操作定位到hashentry链表，然后遍历链表就可以完成相应的操作了。
能够定位元素以后concurrenthashmap就已经具有了hashmap的功能了，现在要解决的就是如何并发的问题。要解决并发问题，加锁是必不可免的。再回头看segment的类图，可以看到segment除了有一个volatile类型的元素大小count外，segment还是集成自reentrantlock的。另外在前面的原子操作和锁机制中介绍过，要想最大限度的支持并发，那么能够利用的思路就是尽量读操作不加锁，写操作不加锁。如果是读操作不加锁，写操作加锁，对于竞争资源来说就需要定义为volatile类型的。volatile类型能够保证happens-before法则，所以volatile能够近似保证正确性的情况下最大程度的降低加锁带来的影响，同时还与写操作的锁不产生冲突。
同时为了防止在遍历hashentry的时候被破坏，那么对于hashentry的数据结构来说，除了value之外其他属性就应该是常量，否则不可避免的会得到concurrentmodificationexception。这就是为什么hashentry数据结构中key,hash,next是常量的原因(final类型）。
有了上面的分析和条件后再来看segment的get/put/remove就容易多了。
get操作
清单2 segment定位元素
v get(object key, int hash) {
if (count != 0) { // read-volatile
hashentry 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;
}
hashentry getfirst(int hash) {
hashentry[] tab = table;
return tab[hash & (tab.length - 1)];
}
v readvalueunderlock(hashentry e) {
lock();
try {
return e.value;
} finally {
unlock();
}
}
清单2 描述的是segment如何定位元素。首先判断segment的大小count>0，segment的大小描述的是hashentry不为空(key不为空)的个数。如果segment中存在元素那么就通过getfirst定位到指定的hashentry链表的头节点上，然后遍历此节点，一旦找到key对应的元素后就返回其对应的值。但是在清单2 中可以看到拿到hashentry的value后还进行了一次判断操作，如果为空还需要加锁再读取一次（readvalueunderlock）。为什么会有这样的操作？尽管concurrenthashmap不允许将value为null的值加入，但现在仍然能够读到一个为空的value就意味着此值对当前线程还不可见（这是因为hashentry还没有完全构造完成就赋值导致的，后面还会谈到此机制）。
put操作
清单3 描述的是segment的put操作。首先就需要加锁了，修改一个竞争资源肯定是要加锁的，这个毫无疑问。需要说明的是segment集成的是reentrantlock，所以这里加的锁也就是独占锁，也就是说同一个segment在同一时刻只有能一个put操作。
接下来来就是检查是否需要扩容，这和hashmap一样，如果需要的话就扩大一倍，同时进行rehash操作。
查找元素就和get操作是一样的，得到元素就直接修改其值就好了。这里onlyifabsent只是为了实现concurrentmap的putifabsent操作而已。需要说明以下几点：
- 如果找到key对于的hashentry后直接修改就好了，如果找不到那么就需要构造一个新的hashentry出来加到hash对于的hashentry的头部，同时就的头部就加到新的头部后面。这是因为hashentry的next是final类型的，所以只能修改头节点才能加元素加入链表中。
- 如果增加了新的操作后，就需要将count 1写回去。前面说过count是volatile类型，而读取操作没有加锁，所以只能把元素真正写回segment中的时候才能修改count值，这个要放到整个操作的最后。
- 在将新的hashentry写入table中时是通过构造函数来设置value值的，这意味对table的赋值可能在设置value之前，也就是说得到了一个半构造完的hashentry。这就是重排序可能引起的问题。所以在读取操作中，一旦读到了一个value为空的value是就需要加锁重新读取一次。为什么要加锁？加锁意味着前一个写操作的锁释放，也就是前一个锁的数据已经完成写完了了，根据happens-before法则，前一个写操作的结果对当前读线程就可见了。当然在jdk 6.0以后不一定存在此问题。
- 在segment中table变量是volatile类型，多次读取volatile类型的开销要不非volatile开销要大，而且编译器也无法优化，所以在put操作中首先建立一个临时变量tab指向table，多次读写tab的效率要比volatile类型的table要高，jvm也能够对此进行优化。
清单3 segment的put操作
v put(k key, int hash, v value, boolean onlyifabsent) {
lock();
try {
int c = count;
if (c > threshold) // ensure capacity
rehash();
hashentry[] tab = table;
int index = hash & (tab.length - 1);
hashentry first = tab[index];
hashentry 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(key, hash, first, value);
count = c; // write-volatile
}
return oldvalue;
} finally {
unlock();
}
}
remove 操作
清单4 描述了segment删除一个元素的过程。同put一样，remove也需要加锁，这是因为对table可能会有变更。由于hashentry的next节点是final类型的，所以一旦删除链表中间一个元素，就需要将删除之前或者之后的元素重新加入新的链表。而segment采用的是将删除元素之前的元素一个个重新加入删除之后的元素之前（也就是链表头结点）来完成新链表的构造。
清单4 segment的remove操作
v remove(object key, int hash, object value) {
lock();
try {
int c = count - 1;
hashentry[] tab = table;
int index = hash & (tab.length - 1);
hashentry first = tab[index];
hashentry 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 newfirst = e.next;
for (hashentry p = first; p != e; p = p.next)
newfirst = new hashentry(p.key, p.hash,
newfirst, p.value);
tab[index] = newfirst;
count = c; // write-volatile
}
}
return oldvalue;
} finally {
unlock();
}
}
下面的示意图描述了如何删除一个已经存在的元素的。假设我们要删除b3元素。首先定位到b3所在的segment，然后再定位到segment的table中的b1元素，也就是bx所在的链表。然后遍历链表找到b3，找到之后就从头结点b1开始构建新的节点b1（蓝色）加到b4的前面，继续b1后面的节点b2构造b2（蓝色），加到由蓝色的b1和b4构成的新的链表。继续下去，直到遇到b3后终止，这样就构造出来一个新的链表b2（蓝色）->b1（蓝色）->b4->b5，然后将此链表的头结点b2（蓝色）设置到segment的table中。这样就完成了元素b3的删除操作。需要说明的是，尽管就的链表仍然存在(b1->b2->b3->b4->b5)，但是由于没有引用指向此链表，所以此链表中无引用的（b1->b2->b3）最终会被gc回收掉。这样做的一个好处是，如果某个读操作在删除时已经定位到了旧的链表上，那么此操作仍然将能读到数据，只不过读取到的是旧数据而已，这在多线程里面是没有问题的。
除了对单个元素操作外，还有对全部的segment的操作，比如size()操作等。
size操作
size操作涉及到统计所有segment的大小，这样就会遍历所有的segment，如果每次加锁就会导致整个map都被锁住了，任何需要锁的操作都将无法进行。这里用到了一个比较巧妙的方案解决此问题。
在segment中有一个变量modcount，用来记录segment结构变更的次数，结构变更包括增加元素和删除元素，每增加一个元素操作就 1，每进行一次删除操作 1，每进行一次清空操作(clear)就 1。也就是说每次涉及到元素个数变更的操作modcount都会 1，而且一直是增大的，不会减小。
遍历两次concurrenthashmap中的segments，每次遍历是记录每一个segment的modcount，比较两次遍历的modcount值的和是否相同，如果相同就返回在遍历过程中获取的segment的count的和，也就是所有元素的个数。如果不相同就重复再做一次。重复一次还不相同就将所有segment锁住，一个一个的获取其大小(count)，最后将这些count加起来得到总的大小。当然了最后需要将锁一一释放。清单5 描述了这个过程。
这里有一个比较高级的话题是为什么在读取modcount的时候总是先要读取count一下。为什么不是先读取modcount然后再读取count的呢？也就是说下面的两条语句能否交换下顺序？
sum = segments[i].count;
mcsum = mc[i] = segments[i].modcount;
答案是不能！为什么？这是因为modcount总是在加锁的情况下才发生变化，所以不会发生多线程同时修改的情况，也就是没必要时volatile类型。另外总是在count修改的情况下修改modcount，而count是一个volatile变量。于是这里就充分利用了volatile的特性。
根据happens-before法则，第（3）条：对volatile字段的写入操作happens-before于每一个后续的同一个字段的读操作。也就是说一个操作c在volatile字段的写操作之后，那么volatile写操作之前的所有操作都对此操作c可见。所以修改modcount总是在修改count之前，也就是说如果读取到了一个count的值，那么在count变化之前的modcount也就能够读取到，换句话说就是如果看到了count值的变化，那么就一定看到了modcount值的变化。而如果上面两条语句交换下顺序就无法保证这个结果一定存在了。
在concurrenthashmap.containsvalue中，可以看到每次遍历segments时都会执行int c = segments[i].count;，但是接下来的语句中又不用此变量c，尽管如此jvm仍然不能将此语句优化掉，因为这是一个volatile字段的读取操作，它保证了一些列操作的happens-before顺序，所以是至关重要的。在这里可以看到：
concurrenthashmap将volatile发挥到了极致！
另外isempty操作于size操作类似，不再累述。
清单5 concurrenthashmap的size操作
public int size() {
final segment[] segments = this.segments;
long sum = 0;
long check = 0;
int[] mc = new int[segments.length];
// try a few times to get accurate count. on failure due to
// continuous async changes in table, resort to locking.
for (int k = 0; k < retries_before_lock; k) {
check = 0;
sum = 0;
int mcsum = 0;
for (int i = 0; i < segments.length; i) {
sum = segments[i].count;
mcsum = mc[i] = segments[i].modcount;
}
if (mcsum != 0) {
for (int i = 0; i < segments.length; i) {
check = segments[i].count;
if (mc[i] != segments[i].modcount) {
check = -1; // force retry
break;
}
}
}
if (check == sum)
break;
}
if (check != sum) { // resort to locking all segments
sum = 0;
for (int i = 0; i < segments.length; i)
segments[i].lock();
for (int i = 0; i < segments.length; i)
sum = segments[i].count;
for (int i = 0; i < segments.length; i)
segments[i].unlock();
}
if (sum > integer.max_value)
return integer.max_value;
else
return (int)sum;
}
concurrentskiplistmap/set
本来打算介绍下concurrentskiplistmap的，结果打开源码一看，彻底放弃了。那里面的数据结构和算法我估计研究一周也未必能够完全弄懂。很久以前我看treemap的时候就头大，想想那些复杂的“红黑二叉树”我头都大了。这些都归咎于从前没有好好学习《数据结构和算法》，现在再回头看这些复杂的算法感觉非常头疼，为了减少脑细胞的死亡，暂且还是不要惹这些“玩意儿”。有兴趣的可以看看 中对treemap的介绍。
参考资料：
- 指令重排序与happens-before法则
©2009-2014 imxylz
|求贤若渴