并发容器分析（五）--ConcurrentHashMap

一、简介

    ConcurrentHashMap是Map的一种并发实现，在该类中元素的read操作都是无锁了，而write操作需要被同步。这非常适合于读操作远大于写操作的情况。在实现过程中，ConcurrentHashMap将所有元素分成了若干个segment，每个segment是独立的，在一个segment上加锁并不影响其他segment的操作。segment本身是一个hashtable，对于一个加入ConcurrentHashMap的<key, value>对，key的hash值中的高位被用来索引segment，而低位用于segment中的索引。

二、segment实现

    segment是ConcurrentHashMap存储元素的基本段，它本身是一个hashtable的实现，read操作时无锁的，write需要同步，定义如下：

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
        implements ConcurrentMap<K, V>, Serializable {


    /**
     *  key, hash, next都是不可改的
   *  value值可被重写
   */
    static final class HashEntry<K,V> {
        final K key;
        final int hash;
        volatile V value;
        final HashEntry<K,V> next;

        ...
    }

    static final class Segment<K,V> extends ReentrantLock implements Serializable {

        transient volatile int count;
        transient volatile HashEntry[] table;
        // 当segment中元素个数达到threshold时，需要rehash
        transient int threshold;
    }

  ...
} 

    segment的read操作：

   static final class Segment<K,V> extends ReentrantLock implements Serializable {

        HashEntry<K,V> getFirst(int hash) {
            HashEntry[] tab = table;
            return (HashEntry<K,V>) tab[hash & (tab.length - 1)];
        }
      
       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当中的key和next都是final的，所以segment之上的操作不可能影响HashEntry列表之间相对的顺序，而value是可变的，当第一次读值失败时，尝试加锁读。

    segment的replace操作：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

       /**
          * replace操作是就地替换，HashEntry的value是非final的
       */
        boolean replace(K key, int hash, V oldValue, V newValue) {
            lock();  // replace操作是同步的
        try {
                // 得到该hash值对应的entry列表
           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();
            }
        }

  ...
}

    segmet的put操作：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

        V put(K key, int hash, V value, boolean onlyIfAbsent) {
            lock(); // put是同步的
        try {
                int c = count;
                if (c++ > threshold) // ensure capacity
                    rehash();
                HashEntry[] tab = table;
                int index = hash & (tab.length - 1);
                HashEntry<K,V> first = (HashEntry<K,V>) 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;
                    // HashEntry的next是只读的，新加入的entry只能放在头部
             tab[index] = new HashEntry<K,V>(key, hash, first, value);
                    count = c; // write-volatile
                }
                return oldValue;
            } finally {
                unlock();
            }
        }

  ...
}

    segment的remove操作一种copy on write 的方法，保留被删元素之后的列表，copy被删元素之前的hashEntry：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

        V remove(Object key, int hash, Object value) {
            lock();
            try {
                int c = count - 1;
                HashEntry[] tab = table;
                int index = hash & (tab.length - 1);
                HashEntry<K,V> first = (HashEntry<K,V>)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)) { // copy on write
                        oldValue = v;
                        ++modCount;
                        // e之后的列表可以保留，只需要重新创建e之前的HashEntry即可
                 HashEntry<K,V> newFirst = e.next;
                        // copy on write e之前的HashEntry
                        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();
            }
        }

  ...
}

    segment的rehash操作实现比较特别，为了保证rehash过程中copy的元素尽可能少，segment在rehash时Entry入口的个数是以2的倍数增长，这可以保证一个entry在rehash之后要么在原来的列表中，要么在下一个列表中：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

        void rehash() {
            // 局部变量引用table
            HashEntry[] oldTable = table;            
            int oldCapacity = oldTable.length;
            if (oldCapacity >= MAXIMUM_CAPACITY)
                return;

            // 右移1位相当于乘以2
            HashEntry[] newTable = new HashEntry[oldCapacity << 1];
            threshold = (int)(newTable.length * loadFactor);
            int sizeMask = newTable.length - 1;
            for (int i = 0; i < oldCapacity ; i++) {
                // 第i个entry列表
           HashEntry<K,V> e = (HashEntry<K,V>)oldTable[i];

                if (e != null) {
                    HashEntry<K,V> next = e.next;
                    // 在新table上的索引
                    int idx = e.hash & sizeMask;

                    if (next == null)
                        newTable[idx] = e;
                    else {
                        // 寻找该entry列表末端，rehash之后idx相同的元素
                        // 这些元素不需要被copy
                        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;
                            }
                        }
                        // 将lastRun之后的整个列表挂到新位置上
                        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 = (HashEntry<K,V>)newTable[k];
                            newTable[k] = new HashEntry<K,V>(p.key, p.hash,
                                                             n, p.value);
                        }
                    }
                }
            }
            table = newTable;
        }

  ...
}

三、ConcurrentHashMap方法实现

    ConcurrentHashMap在Segment的基础上，通过首先将<key, value>对hash到一个segment，再由segment实现对entry的管理。

    ConcurrentHashMap的get实现：

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
        implements ConcurrentMap<K, V>, Serializable {

    final Segment<K,V> segmentFor(int hash) {
        return (Segment<K,V>) segments[(hash >>> segmentShift) & segmentMask];
    }

    public V get(Object key) {
        int hash = hash(key); // throws NullPointerException if key null
        return segmentFor(hash).get(key, hash);
    }

  ...
}

    ConcurrentHashMap的put和get方法：

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
        implements ConcurrentMap<K, V>, Serializable {

    public V put(K key, V value) {
        if (value == null)
            throw new NullPointerException();
        int hash = hash(key);
        return segmentFor(hash).put(key, hash, value, false);
    }

    public V remove(Object key) {
        int hash = hash(key);
        return segmentFor(hash).remove(key, hash, null);
    }

  ...
}

 

 

评论

1 楼 ykdsg 2011-03-17  

我想问下rehash（）方法中 // 将lastRun之后的整个列表挂到新位置上    newTable[lastIdx] = lastRun;  这个怎么保证newTable[lastIdx]这个位置就是空值