java线程基础

ConcurrentHashMap是HashMap的线程安全版本，ConcurrentSkipListMap是TreeMap的线程安全版本

在HashMap中首先由一个对象数组table是不可避免的，修饰符transient只是表示序列号的时候不被存储而已。size描述的是Map中元素的大小，threshold描述的是达到指定元素个数后需要扩容，loadFactor是扩容因子(loadFactor>0)，也就是计算threshold的。那么元素的容量就是table.length，也就是数组的大小。换句话说，如果存取的元素大小达到了整个容量(table.length)的loadFactor倍（也就是table.length*loadFactor个），那么就需要扩充容量了。在HashMap中每次扩容就是将扩大数组的一倍，使数组大小为原来的两倍。

然后接下来看如何将一个元素映射到数组table中。显然要映射的key是一个无尽的超大集合，而table是一个较小的有限集合，那么一种方式就是将key编码后的hashCode值取模映射到table上，这样看起来不错。但是在Java中采用了一种更高效的办法。由于与(&)是比取模(%)更高效的操作，因此Java中采用hash值与数组大小-1后取与来确定数组索引的。为什么这样做是更有效的？参考资料7对这一块进行非常详细的分析，这篇文章的作者非常认真，也非常仔细的分析了里面包含的思想。

清单1 indexFor片段

static int indexFor(int h, int length) {
    return h & (length-1);
}

前面说明，既然是大集合映射到小集合上，那么就必然存在“碰撞”，也就是不同的key映射到了相同的元素上。那么HashMap是怎么解决这个问题的？

在HashMap中采用了下面方式，解决了此问题。

1. 同一个索引的数组元素组成一个链表，查找允许时循环链表找到需要的元素。
2. 尽可能的将元素均匀的分布在数组上。

 

对于问题1，HashMap采用了上图的一种数据结构。table中每一个元素是一个Map.Entry，其中Entry包含了四个数据，key,value,hash,next。key和value是存储的数据；hash是元素key的Hash后的表现形式（最终要映射到数组上），这里链表上所有元素的hash经过清单1 的indexFor后将得到相同的数组索引；next是指向下一个元素的索引，同一个链表上的元素就是通过next串联起来的。

再来看问题2 尽可能的将元素均匀的分布在数组上这个问题是怎么解决的。首先清单2 是将key的hashCode经过一系列的变换，使之更符合小数据集合的散列模型。

清单2 hashCode的二次散列

static int hash(int h) {
    // This function ensures that hashCodes that differ only by
    // constant multiples at each bit position have a bounded
    // number of collisions (approximately 8 at default load factor).
    h ^= (h >>> 20) ^ (h >>> 12);
    return h ^ (h >>> 7) ^ (h >>> 4);
}

至于清单2 为什么这样散列我没有找到依据，也没有什么好的参考资料。参考资料1 分析了此过程，认为是一种比较有效的方式，有兴趣的可以研究下。

第二点就是在清单1 的描述中，尽可能的与数组的长度减1的数与操作，使之分布均匀。这在参考资料7 中有介绍。

第三点就是构造数组时数组的长度是2的倍数。清单3 反映了这个过程。为什么要是2的倍数？在参考资料7 中分析说是使元素尽可能的分布均匀。

清单3 HashMap 构造数组

// Find a power of 2 >= initialCapacity
int capacity = 1;
while (capacity < initialCapacity)
    capacity <<= 1;

this.loadFactor = loadFactor;
threshold = (int)(capacity * loadFactor);
table = new Entry[capacity];

另外loadFactor的默认值0.75和capacity的默认值16是经过大量的统计分析得出的，很久以前我见过相关的数据分析，现在找不到了，有兴趣的可以查询相关资料。这里不再叙述了。

有了上述原理后再来分析HashMap的各种方法就不是什么问题的。

清单4 HashMap的get操作

public V get(Object key) {
    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;
}

清单4 描述的是HashMap的get操作，在这个操作中首先判断key是否为空，因为为空的话总是映射到table的第0个元素上（可以看上面的清单2和清单1）。然后就需要查找table的索引。一旦找到对应的Map.Entry元素后就开始遍历此链表。由于不同的hash可能映射到同一个table[index]上，而相同的key却同时映射到相同的hash上，所以一个key和Entry对应的条件就是hash(key)==e.hash 并且key.equals(e.key)。从这里我们看到，Object.hashCode()只是为了将相同的元素映射到相同的链表上（Map.Entry)，而Object.equals()才是比较两个元素是否相同的关键！这就是为什么总是成对覆盖hashCode()和equals()的原因。

默认情况下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，有兴趣的可以参考资料1和参考资料2。总之它的目的就是使元素能够均匀的分布在不同的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<K,V> 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<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;
}
HashEntry<K,V> getFirst(int hash) {
    HashEntry<K,V>[] tab = table;
    return tab[hash & (tab.length - 1)];
}

V readValueUnderLock(HashEntry<K,V> 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<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();
    }
}

 

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<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();
    }
}

下面的示意图描述了如何删除一个已经存在的元素的。假设我们要删除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<K,V>[] 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;
}