本文旨在探讨在使用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<Integer, Integer> 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()方法。
下面是使用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<Integer, Integer> 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<Object>> tasks = Collections.nCopies(4, Executors.callable(runnable)); // invokeAll会阻塞，直到所有任务完成 List<Future<Object>> futures = executorService.invokeAll(tasks); // 此时所有任务已完成，可以安全地检查map的大小 System.out.println("Map size: " + map.size()); // 预期输出 1000 } catch (InterruptedException e) { Thread.currentThread().interrupt(); System.err.println("任务执行被中断: " + e.getMessage()); } finally { executorService.shutdown(); // 关闭线程池 // 建议等待线程池终止，确保所有资源释放 if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) { System.err.println("线程池未在指定时间内终止。"); } } } }
3. 代码解析与注意事项
-
ExecutorService.invokeAll(Collection extends Callable
> tasks) :- 该方法提交一个Callable任务集合，并等待所有任务完成。
- 它返回一个Future列表，与提交的任务顺序一致，每个Future代表对应任务的执行结果。
- 此方法是阻塞的，直到所有任务都已完成（正常完成、异常终止或被取消）才返回。这正是我们确保所有put操作完成的关键。
-
Executors.callable(Runnable task):
- 这是一个便捷工具方法，用于将一个Runnable实例包装成一个Callable实例。
- 由于Runnable的run()方法没有返回值，因此转换后的Callable的泛型类型为Object（或Void），其call()方法返回null。
-
Collections.nCopies(int n, T o):
- 此方法返回一个包含n个相同元素o的不可变列表。
- 在这里，它用于快速创建4个相同的Callable任务，省去了手动循环添加的步骤。
-
finally 块中的资源管理:
- executorService.shutdown()：此方法启动线程池的有序关闭，不再接受新任务，但会完成已提交的任务。
- executorService.awaitTermination(long timeout, TimeUnit unit)：这是一个重要的步骤，它会阻塞当前线程，直到所有任务执行完毕且线程池终止，或者达到指定的超时时间。这确保了在程序退出前，所有后台线程都已清理，避免资源泄露或不确定行为。在测试场景下，如果不调用awaitTermination，主线程可能会在后台任务完成前退出JVM。
4. 总结
本教程详细阐述了在并发环境下测试ConcurrentHashMap时，由于线程执行的异步性可能导致观测结果不准确的问题。核心在于，ConcurrentHashMap保证了内部操作的线程安全，但外部对操作完成时机的感知需要额外的同步机制。通过使用ExecutorService.invokeAll()方法，我们可以有效地确保所有并发任务都已执行完毕，从而获得对ConcurrentHashMap状态的准确观测。这对于编写健壮的并发测试代码和理解并发编程中的时序问题至关重要。正确地管理线程生命周期和任务完成状态，是构建可靠并发应用的基础。
评论（已关闭）
评论已关闭