Java避坑案例 - ConcurrentHashMap 的使用陷阱

Pre

J.U.C Review - 并发容器集合解析

Java Review - 并发组件ConcurrentHashMap使用时的注意事项及源码分析

概述

JDK 1.5 后推出的 ConcurrentHashMap，是一个高性能的线程安全的哈希表容器。“线程安全”这四个字特别容易让人误解，因为 ConcurrentHashMap 只能保证提供的原子性读写操作是线程安全的.

我们通常误以为使用了 ConcurrentHashMap 后，就不需要担心线程安全问题。然而ConcurrentHashMap 只能保证单个操作（如 put、get）的线程安全，无法确保多个操作（如 size 和 putAll）之间的一致性。

场景

有一个含 900 个元素的 Map，现在再补充 100 个元素进去，这个补充操作由 10 个线程并发进行。

问题复现

private static int THREAD_COUNT = 10;
private static int ITEM_COUNT = 1000;

private ConcurrentHashMap<String, Long> getData(int count) {
    return LongStream.rangeClosed(1, count)
        .boxed()
        .collect(Collectors.toConcurrentMap(
            i -> UUID.randomUUID().toString(),
            i -> i,
            (o1, o2) -> o1,
            ConcurrentHashMap::new
        ));
}

@GetMapping("wrongVersion")
public String wrongVersion() throws InterruptedException {
    // 初始化900个元素
    ConcurrentHashMap<String, Long> concurrentHashMap = getData(ITEM_COUNT - 100);
    log.info("init size: {}", concurrentHashMap.size());

    // 使用 ForkJoinPool 进行并发操作
    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    forkJoinPool.execute(() -> IntStream.rangeClosed(1, 10).parallel().forEach(i -> {
        // 查询缺少的元素数量
        int gap = ITEM_COUNT - concurrentHashMap.size();
        log.info("gap size: {}", gap);

        // 补充缺少的元素
        concurrentHashMap.putAll(getData(gap));
    }));

    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);

    // 输出最终大小
    log.info("finish size: {}", concurrentHashMap.size());
    return "OK";
}

开发人员误以为使用了 ConcurrentHashMap 就不会有线程安全问题，于是不加思索地写出了下面的代码：在每一个线程的代码逻辑中先通过 size 方法拿到当前元素数量，计算 ConcurrentHashMap 目前还需要补充多少元素，并在日志中输出了这个值，然后通过
putAll 方法把缺少的元素添加进去。

现象

1. 初始大小：900
2. 多个线程执行时计算到的差值（gap）不一致，有的线程甚至发现了负值。
3. 最终 ConcurrentHashMap 的大小不是预期的 1000。

问题分析

1. ConcurrentHashMap 的误用：
   虽然 ConcurrentHashMap 能保证基本的线程安全，但无法假设多个方法调用之间的一致性。例如，当一个线程通过 size() 获取元素个数时，另一个线程可能已经向 Map 中插入了新元素。这会导致不同线程看到的数据不一致。

2. 非原子性操作的问题：

   • size() 返回的是 ConcurrentHashMap 的瞬时快照，但在并发情况下，它可能只是一个中间状态，不能用于精确控制逻辑。
   • putAll() 并不是原子操作，如果在它执行的过程中其他线程也在修改 Map，就会出现预期外的结果。

解决方案：加锁版本

为了解决上述问题，我们可以在 补充元素的逻辑 上加锁，确保这部分逻辑的原子性。

Fix

@GetMapping("rightVersion")
public String rightVersion() throws InterruptedException {
    // 初始化900个元素
    ConcurrentHashMap<String, Long> concurrentHashMap = getData(ITEM_COUNT - 100);
    log.info("init size: {}", concurrentHashMap.size());

    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    forkJoinPool.execute(() -> IntStream.rangeClosed(1, 10).parallel().forEach(i -> {
        synchronized (concurrentHashMap) {
            // 查询缺少的元素数量
            int gap = ITEM_COUNT - concurrentHashMap.size();
            log.info("gap size: {}", gap);

            // 补充缺少的元素
            concurrentHashMap.putAll(getData(gap));
        }
    }));

    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);

    log.info("finish size: {}", concurrentHashMap.size());
    return "OK";
}

效果

1. 只有一个线程会发现需要补充 100 个元素，其他线程看到的 gap 为 0。
2. 最终 ConcurrentHashMap 的大小为 1000，符合预期。

改进与更优方案：使用原子操作避免加锁

虽然加锁能解决问题，但这样会牺牲并发性能。我们可以通过一些更优的方案来避免锁的使用：

使用 computeIfAbsent 等原子方法

ConcurrentHashMap 提供了如 computeIfAbsent、merge 等原子操作，我们可以利用这些方法简化并发逻辑。

单线程执行填充操作

由于填充操作涉及多个步骤，不易实现原子性。可以考虑将所有填充操作交给一个线程来完成，从而避免并发问题。

@GetMapping("betterVersion")
public String betterVersion() throws InterruptedException {
    ConcurrentHashMap<String, Long> concurrentHashMap = getData(ITEM_COUNT - 100);
    log.info("init size: {}", concurrentHashMap.size());

    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    // 使用单线程来执行填充操作，避免并发问题
    forkJoinPool.submit(() -> {
        int gap = ITEM_COUNT - concurrentHashMap.size();
        log.info("gap size: {}", gap);
        concurrentHashMap.putAll(getData(gap));
    }).get(); // 等待任务完成

    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);

    log.info("finish size: {}", concurrentHashMap.size());
    return "OK";
}

总结

使用并发工具类的注意事项

1. 多操作间状态一致性问题：
   ConcurrentHashMap 能保证单次操作的线程安全，但无法保证多个操作之间的一致性。

2. 避免使用快照方法控制流程：
   在并发情况下，size() 等方法只能用作参考，不能用于控制流程逻辑。

3. 尽量使用原子操作：
   使用 computeIfAbsent、merge 等方法，可以在一定程度上避免显式加锁。

4. 不要滥用锁：
   在高并发场景下，频繁加锁会严重影响性能，合理设计并发逻辑才能充分发挥 ConcurrentHashMap 的性能优势。

教训： 误以为使用了并发工具就可以解决一切线程安全问题，期望通过把线程不安全的类替换为线程安全的类来一键解决问题。比如，认为使用了 ConcurrentHashMap 就可以解决线程安全问题，没对复合逻辑加锁导致业务逻辑错误。如果希望在一整段业务逻辑中，对容器的操作都保持整体一致性的话，需要加锁处理