文章目录

ConcurrentHashMap的使用介绍和底层原理解析和开源框架的使用实例
- 1. ConcurrentHashMap介绍
- 2. ConcurrentHashMap底层原理
- 3. ConcurrentHashMap主要方法- put(K key, V value):添加元素。
- 4. 总结
- 5. 框架中的应用
- 6. 操作技巧
- 7. 运维部署(生产环境注意事项)
- 8. ConcurrentHashMap扩展-JDK8改进
- 9. 总结

ConcurrentHashMap的使用介绍和底层原理解析和开源框架的使用实例

1. ConcurrentHashMap介绍

ConcurrentHashMap是JDK1.5提供的线程安全的HashMap,它允许多个线程并发访问哈希表,并发修改map中的数据而不会产生死锁。ConcurrentHashMap适用于高并发的环境下,可以替代synchronized实现的同步HashMap。ConcurrentHashMap的并发度很高,吞吐量也很高。

2. ConcurrentHashMap底层原理

ConcurrentHashMap底层采用“分段锁”机制,将数据分成一段段的存储,然后给每一段数据配一把锁,当一个线程占用锁访问其中一个段数据时,其他段的数据也能被其他线程访问,

能够实现真正的并发访问。ConcurrentHashMap结构如下:

Segment数组:存放数据段,默认16个段,数组大小始终为2的幂次方。
每个Segment是一把锁,锁定一个段数据的所有访问。
每个Segment包含一个HashEntry数组,用来存储链表结构的数据。
一个HashEntry就是一个节点,存储key-value键值对。

3. ConcurrentHashMap主要方法- put(K key, V value):添加元素。

get(K key):获取元素。
size():返回ConcurrentHashMap的大小。
isEmpty():判断ConcurrentHashMap是否为空。这些方法都可以在多线程环境下调用,方法内部会处理线程安全问题。示例代码:

java">ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

 // 添加元素
 map.put("a", 1);
 map.put("b", 2);
 map.put("c", 3);

 // 获取元素
 Integer a = map.get("a");
 Integer b = map.get("b");
 Integer c = map.get("c");

 // 大小和判断是否为空
 int size = map.size(); 
 boolean empty = map.isEmpty();

4. 总结

ConcurrentHashMap通过锁分段技术,实现高度的并发访问,大大提高了HashMap的吞吐量,是高并发环境下一个很好的选择。理解ConcurrentHashMap的原理和结构,可以更好的发挥其高性能特点。

5. 框架中的应用

ConcurrentHashMap在很多开源框架中广泛应用,这里举两个例子:

Spring Cache 注解 @Cacheable 的底层缓存存储就是采用ConcurrentHashMap来实现的。Spring Cache 对象存储在 ConcurrentHashMap<name, Cache> 中,name为缓存名称。
Mybatis映射 SqlSessionFactory 里面的Configuration对象的mappedStatements属性就是一个ConcurrentHashMap。它的key是statement id, value是封装好的映射语句MappedStatement对象。

这两个例子都采用ConcurrentHashMap来存放数据,体现了它的 thread-safe 特性,可以在高并发场景下安全地操作数据。

6. 操作技巧

在开发中,我们也要注意ConcurrentHashMap的一些操作技巧:

初始化大小最好是2的幂次方,默认是16,可以根据实际数据量选择合适大小,此可以减少rehash的次数,提高效率。
如果需要Iterator遍历,最好使用entrySet来遍历Map。因为如果在遍历的过程中,Map的数据发生了变化(增加、删除元素),迭代器并不会抛出ConcurrentModificationException异常。
在计算ConcurrentHashMap的size()时,如果此时有其他线程正在进行添加/删除操作,计算出的size值可能是不准确的。如果需要精确的size值,可使用mappingCount()方法。
如果希望ConcurrentHashMap中的key或value组成固定顺序,可以使用TreeMap。ConcurrentHashMap的Key-Value是无序的。
在使用ConcurrentHashMap的过程中,如果遇到元素添加或删除较慢的情况,应考虑map的容量是否过小,是否需要扩容。扩容会带来性能消耗。

示例代码:

java">// 初始化大小
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>(32);

// entrySet遍历 
for (Map.Entry<String, String> entry : map.entrySet()) {
    String key = entry.getKey();
    String value = entry.getValue();
    // do something
}

// mappingCount()计算精确大小
int size = map.mappingCount();

7. 运维部署(生产环境注意事项)

在实际项目中,ConcurrentHashMap的使用也需要考虑一些运维方面的内容:

监控ConcurrentHashMap的大小,避免OOM。ConcurrentHashMap容量过大会导致OOM,需要监控map的size,一旦超过阈值需要考虑清理旧数据或扩容map。
关注CPU使用率和负载。ConcurrentHashMap高并发会导致CPUUsage和负载升高,需要及时监控和调优。可以通过调大初始容量、扩容更加缓慢、reduce锁粒度等手段优化。
GC频率监控。高并发下,ConcurrentHashMap会产生大量临时对象,导致GC频繁,GC时间长会影响系统性能,需要关注老年代GC时间和频率,必要时进行GC优化。
问题诊断工具。ConcurrentHashMap底层采用“分段锁”机制,如果使用不当可以产生死锁。需要熟悉如jstack等诊断工具,及时发现死锁问题并解决。
负载均衡。在高并发下,如果ConcurrentHashMapbottleneck,需要考虑尽量分散压力,可以采取加机器、分布式相关手段进行负载均衡。
测试Verification。高并发场景下ConcurrentHashMap性能表现和同步Map相比会更加复杂,需要进行充分的性能测试,判断是否达到预期效果。修改ConcurrentHashMap的capacity、segment数、rehash等策略都需要充分评估和验证。

这些运维方面内容,可以让ConcurrentHashMap在生产环境中运行更加稳定可靠。总之,ConcurrentHashMap是JDK提供的高性能Map实现,但在实际生产环境中,依然需要运维团队投入大量时间去监控、诊断和优化才能发挥其最高性能。

8. ConcurrentHashMap扩展-JDK8改进

在JDK8中,ConcurrentHashMap进行了较大改进,比较重要的有两点:

采用CAS操作替换重量级锁,降低锁粒度,实现更高的并发度。
采用红黑树替换链表,提高查询效率。具体改进如下:

Segment改为Node,每个Node是一个链表结构的首节点。不再分段锁定,采用CAS操作同步机制。
采用 volatile + CAS 操作线程安全地修改节点,代替重量级锁。
链表长度超过8自动转换为红黑树,提高查询效率。节点采用二叉查找树结构。
size属性和mappingCount方法删除,采用遍历计数的方式统计大小。
put方法不再加锁,采用CAS操作,删除“死循环”逻辑。
初始化大小和扩容机制改进。默认大小为16,扩容为原来的2倍。

改进代码示例:

java">static final int MIN_TREEIFY_CAPACITY = 64;  //链表转换红黑树阈值

/**
 * Initializes or doubles table size.  If null, allocates in
 * accord with initial capacity target held in field.  Otherwise,
 * because we are using power-of-two expansion, the elements from
 * each bin must either stay at same index, or move with a power
 * of two offset in the new table.
 *
 * @return the table
 */
final Node<K,V>[] resize() {
    Node<K,V>[] oldTab = table;
    int oldCap = (oldTab == null) ? 0 : oldTab.length;
    int oldThr = threshold;
    int newCap, newThr = 0;
    if (oldCap > 0) {
        // 超过最大值就不再扩充了,就只好随你碰撞去吧
        if (oldCap >= MAXIMUM_CAPACITY) {
            threshold = Integer.MAX_VALUE;
            return oldTab;
        }
        // 没超过最大值,就扩充为原来的2倍
        else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY &&
                 oldCap >= MIN_TREEIFY_CAPACITY)
            newThr = oldThr << 1; // double threshold
    }
    else if (oldThr > 0) // initial capacity was placed in threshold
        newCap = oldThr;
    else {               // zero initial threshold signifies using defaults
        newCap = DEFAULT_INITIAL_CAPACITY;
        newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY);
    }
    // 计算新的resize上限
    if (newThr == 0) {
        float ft = (float)newCap * loadFactor;
        newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ?
                  (int)ft : Integer.MAX_VALUE);
    }
    threshold = newThr;
    @SuppressWarnings({"rawtypes","unchecked"})
        Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
    table = newTab;
    ......
}