MyException - 我的异常网
当前位置:我的异常网» 编程 » Java 并发聚合ConcurrentHashMap

Java 并发聚合ConcurrentHashMap

www.MyException.Cn  网友分享于:2013-02-16  浏览:21次
Java 并发集合ConcurrentHashMap

ConcurrentHashMap是JDK1.5并发包中提供的线程安全的HashMap的实现,其包结构关系如下:

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
        implements ConcurrentMap<K, V>, Serializable {
}
public abstract class AbstractMap<K,V> implements Map<K,V> {
}
public interface ConcurrentMap<K, V> extends Map<K, V> {
}

 ConcurrentHashMap实现并发是通过“锁分离”技术来实现的,也就是将锁拆分,不同的元素拥有不同的锁,ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分,其中每一个片段是一个类似于HashMap的结构,它有一个HashEntry的数组,数组的每一项又是一个链表,通过HashEntry的next引用串联起来,它们有自己的锁。

final Segment<K,V>[] segments;

 Segment继承自ReentrantLock,在创建Segment对象时,其所做的动作就是创建一个指定大小为cap的HashEntry对象数组,并基于数组的大小及loadFactor计算threshold的值:threshold = (int)(newTable.length * loadFactor);

        Segment(int initialCapacity, float lf) {
            loadFactor = lf;
            setTable(HashEntry.<K,V>newArray(initialCapacity));
        }
        void setTable(HashEntry<K,V>[] newTable) {
            threshold = (int)(newTable.length * loadFactor);
            table = newTable;
        }

 

构造函数

public ConcurrentHashMap(int initialCapacity,
                         float loadFactor,
                         int concurrencyLevel)创建一个带有指定初始容量、加载因子和并发级别的新的空映射。 

参数:
initialCapacity - 初始容量。该实现执行内部大小调整,以容纳这些元素。
loadFactor - 加载因子阈值,用来控制重新调整大小。在每 bin 中的平均元素数大于此阈值时,可能要重新调整大小。
concurrencyLevel - 当前更新线程的估计数。该实现将执行内部大小调整,以尽量容纳这些线程。 
抛出: 
IllegalArgumentException - 如果初始容量为负,或者加载因子或 concurrencyLevel 为非正。

    public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();

        if (concurrencyLevel > MAX_SEGMENTS)
            concurrencyLevel = MAX_SEGMENTS;

        // Find power-of-two sizes best matching arguments
        int sshift = 0;
        int ssize = 1;
        while (ssize < concurrencyLevel) {
            ++sshift;
            ssize <<= 1;
        }
        segmentShift = 32 - sshift;
        segmentMask = ssize - 1;
        this.segments = Segment.newArray(ssize);

        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        int c = initialCapacity / ssize;
        if (c * ssize < initialCapacity)
            ++c;
        int cap = 1;
        while (cap < c)
            cap <<= 1;

        for (int i = 0; i < this.segments.length; ++i)
            this.segments[i] = new Segment<K,V>(cap, loadFactor);
    }

 基于如下方法计算ssize的大小:

        int sshift = 0;
        int ssize = 1;
        while (ssize < concurrencyLevel) {
            ++sshift;
            ssize <<= 1;
        }

 默认情况下构造函数的三个值分别为16、0.75f、16。在concurrencyLevel为16的情况下,计算出的ssize值为16,并使用该值作为参数传入Senment的newArray方法创建一个大小为16的Segment对象数组,也就是默认情况下ConcurrentHashMap是用了16个类似HashMap 的结构。

采用下面方法计算cap变量的值:

        int c = initialCapacity / ssize;
        if (c * ssize < initialCapacity)
            ++c;
        int cap = 1;
        while (cap < c)
            cap <<= 1;

 算出的cap为1。

 

 put(Object key,Object value)方法

 ConcurrentHashMap的put方法并没有加synchronized来保证线程同步,而是在Segment中实现同步,如下:

 

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

//下面为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();
            }
        }
 ConcurrentHashMap不能保存value为null值,否则抛出NullPointerException,key也不能为空:

 

 

int hash = hash(key.hashCode());
在HashMap中,null可以作为key也可以为value。和HashMap一样,首先对key.hashCode()进行hash操作,得到key的hash值,然后再根据hash值得到其对应数组的Segment对象,接着调用Segment对象的put方法来完成操作。当调用Segment对象的put方法时,先进行lock操作,接着判断当前存储的对象个数加1后是否大于threshold,如大于则将当前的HashEntry对象数组大小扩大两倍,并将之前存储的对象重新hash转移到新的对象数组中。接下去的动作和HashMap基本一样,通过对hash值和对象数组大小减1进行按位与操作后,找到当前key要存放的数组的位置,接着寻找对应位置上的HashEntry对象链表是否有key、hash值和当前key相同的,如果有则覆盖其value,如果没有则创建一个新的HashEntry对象,赋值给对应位置的数组对象,构成链表。
从上面可以看出ConcurrentHashMap基于concurrencyLevel划分出多个Segment来存储对象,从则避免每次put操作都锁住得锁住整个数组。在默认的情况下可以充许16个线程并发无阻塞的操作集合对象。
remove(Object key)方法:
    public V remove(Object key) {
	int hash = hash(key.hashCode());
        return segmentFor(hash).remove(key, hash, null);
    }
 首先对key进行hash求值,再根据hash值找到对应的Segment对象,调用其remove方法完成删除操作。remove方法跟put方法一样,也是在Segment对象中的方法才加锁。
get(Object key)方法:
 跟put和remove方法一样,首先对key进行hash,再根据该hash值找到对应的Segment对象,然后调用该Segment对象的get方法完成操作。
    public V get(Object key) {
        int hash = hash(key.hashCode());
        return segmentFor(hash).get(key, hash);
    }
Segment的get方法先判断当前HashEntry对象数组的长度是否为0,如果为0则直接返回null。然后用hash值和对象数组长度减1按位与操作得到该位置上的HashEntry对象,然后再遍历该HashEntry对象,如果value不为空,则直接返回value,如果为null,则调用readValueUnderLock()方法取得value并返回,下面方法为Segment的get方法:
        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;
        }
        V readValueUnderLock(HashEntry<K,V> e) {
            lock();
            try {
                return e.value;
            } finally {
                unlock();
            }
        }
 从上面可以看出,ConcurrentHashMap的get方法仅在找到value为null时才加锁,其它情况下都不加锁。
get方法首先通过hash值和HashEntry对象数组大小减1按位与来获取对应位置的HashEntry,在这个步骤中可能因为数组大小的改变而导致获取HashEntry数组对象位置出错,ConcurrentHashMap通过把HashEntry数组对象定义为volatile类型来保证线程同步。
        transient volatile HashEntry<K,V>[] table;
        HashEntry<K,V> getFirst(int hash) {
            HashEntry<K,V>[] tab = table;
            return tab[hash & (tab.length - 1)];
        }
在获取到HashEntry对象后,怎么保证它及其next属性构成的链表上的对象不会改变呢?ConcurrentHashMap中是把HashEntry对象中的hash、key、以及next属性都是final的,也意味着没办法插入一个HashEntry对象到HashEntry基于next属性构成的链表中间或末尾(与HashMap一样,新插入的对象也是插入到HashEntry的表头)。
 和HashMap性能比较
在单线程情况下,ConcurrentHashMap比HashMap性能稍微差一点,在多线程情况下,随着线程数量的增加,ConcurrentHashMap性能明显比HashMap提升,特别是查找性能,而且随着线程数量的增加,ConcurrentHashMap性能并没有出现下降的情况,所以在并发的场景中,使用ConcurrentHashMap比使用HashMap是更好的选择。
 

文章评论

软件开发程序错误异常ExceptionCopyright © 2009-2015 MyException 版权所有