确保并发操作结果可观测性：理解ConcurrentHashMap与线程同步

本文旨在探讨在使用ConcurrentHashMap进行并发写入操作时，如何确保最终结果的准确性与可观测性。我们将分析在多线程环境下，直接检查ConcurrentHashMap大小可能出现不符合预期的原因，并详细介绍如何利用ExecutorService.invokeAll()方法，确保所有并发任务执行完毕后再进行结果验证，从而获得正确且一致的映射大小。

1. 问题现象与分析

在Java并发编程中，ConcurrentHashMap是线程安全的哈希映射实现，它允许在多线程环境下进行高效的并发读写操作，而无需外部同步。然而，在某些测试场景下，即使向ConcurrentHashMap中并发插入数据，最终获取到的映射大小（map.size()）也可能与预期不符。

考虑以下示例代码，旨在通过四个线程向ConcurrentHashMap中各插入1000个条目：

import java.util.Collections;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class ConcurrentMapTest {    public static void main(String[] args) throws InterruptedException {        Map map = new ConcurrentHashMap();        Runnable runnable = () -> {            for (int i = 0; i < 1000; i++) {                map.put(i, i); // 键值对为(0,0)到(999,999)            }        };        ExecutorService executorService = Executors.newFixedThreadPool(4);        for (int i = 0; i < 4; i++) {            executorService.submit(runnable);        }        // 尝试等待一段时间，但无法保证所有任务完成        // TimeUnit.SECONDS.sleep(1);         System.out.println("Map size: " + map.size());        executorService.shutdown(); // 关闭线程池    }}

运行上述代码，我们可能会发现map.size()的输出值并非预期的1000。这是因为尽管ConcurrentHashMap的put操作是线程安全的，但ExecutorService.submit()方法是非阻塞的。这意味着在调用submit后，主线程会立即继续执行下一行代码，即System.out.println(map.size())，而此时提交的四个任务可能尚未全部完成，甚至可能一个都未完成。因此，map.size()反映的是一个瞬时状态，而不是所有任务执行完毕后的最终状态。

问题的核心在于：ConcurrentHashMap保证了其内部状态在并发操作下的正确性，但它不负责外部线程对操作完成时机的同步。我们需要一种机制来等待所有提交的任务完成。

2. 解决方案：利用 ExecutorService.invokeAll()

为了确保在检查ConcurrentHashMap大小之前，所有并发写入任务都已完成，我们可以使用ExecutorService.invokeAll()方法。invokeAll()是一个阻塞方法，它会等待所有提交的Callable任务执行完毕后才返回。

invokeAll()方法接受一个Callable任务集合作为参数，并返回一个Future列表，其中包含了每个任务的执行结果和状态。即使我们不关心任务的返回值，它的阻塞特性也正是我们在此场景下所需要的。

由于invokeAll()期望的是Callable类型的任务，而我们原先使用的是Runnable，因此需要借助Executors.callable()工具方法将Runnable转换为Callable。此外，为了方便地创建多个相同的任务，我们可以利用Collections.nCopies()方法。

火龙果写作

用火龙果，轻松写作，通过校对、改写、扩展等功能实现高质量内容生产。

106 查看详情

下面是使用invokeAll()修正后的代码示例：

import java.util.Collections;import java.util.List;import java.util.Map;import java.util.concurrent.Callable;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Future;public class ConcurrentMapCorrectTest {    public static void main(String[] args) throws InterruptedException {        Map map = new ConcurrentHashMap();        Runnable runnable = () -> {            for (int i = 0; i < 1000; i++) {                // 每个线程都尝试插入键值对 (0,0) 到 (999,999)                // 由于键是固定的，最终map中只会保留每个键的一个值                // 预期结果是1000个不同的键                map.put(i, i);             }        };        ExecutorService executorService = Executors.newFixedThreadPool(4);        try {            // 将Runnable转换为Callable，并创建4个相同的Callable任务            List<Callable