Appearance
源码:labs/java/runtime-concurrency/map-concurrency-compare/src/MapConcurrencyCompare.java
- 原始路径:
labs/java/runtime-concurrency/map-concurrency-compare/src/MapConcurrencyCompare.java - 类型:
java
java
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 对比 HashMap / synchronizedMap / ConcurrentHashMap 的线程安全边界,
* 并演示「get-then-put」复合逻辑在 ConcurrentHashMap 上仍会重复初始化。
*
* 运行:
* cd labs/java/runtime-concurrency/map-concurrency-compare
* javac -d out src/MapConcurrencyCompare.java
* java -cp out MapConcurrencyCompare
*/
public class MapConcurrencyCompare {
private static final String CACHE_KEY = "shared-singleton";
private static final int THREAD_COUNT = 40;
public static void main(String[] args) throws Exception {
System.out.println("====== 1. HashMap 并发写入(非线程安全,结果不可预期)======");
demoHashMapConcurrentPut();
System.out.println("\n====== 2. ConcurrentHashMap:错误写法 get-then-put(可能重复创建)======");
int badCreations = demoGetThenPutBadPattern();
System.out.println("错误写法实际创建次数: " + badCreations + " (期望 1,常 > 1)");
System.out.println("\n====== 3. ConcurrentHashMap:正确写法 computeIfAbsent ======");
int goodCreations = demoComputeIfAbsent();
System.out.println("computeIfAbsent 创建次数: " + goodCreations + " (恒为 1)");
System.out.println("\n====== 4. Collections.synchronizedMap vs ConcurrentHashMap ======");
demoSynchronizedMapContention();
}
/**
* HashMap 在并发 put 下 size 可能小于期望值(丢更新),JDK 高版本还可能结构损坏。
* 本实验用「最终 size」说明:共享字典不要用 HashMap。
*/
static void demoHashMapConcurrentPut() throws InterruptedException {
Map<Integer, Integer> map = new HashMap<>();
int threads = 20;
int keysPerThread = 500;
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch latch = new CountDownLatch(threads);
for (int t = 0; t < threads; t++) {
final int threadId = t;
pool.execute(() -> {
for (int i = 0; i < keysPerThread; i++) {
int key = threadId * keysPerThread + i;
map.put(key, key);
}
latch.countDown();
});
}
latch.await();
pool.shutdown();
int expected = threads * keysPerThread;
System.out.println("期望条目数: " + expected + " | HashMap 实际 size: " + map.size()
+ " (若小于期望值,说明并发 put 已丢数据)");
/*
* 静态阅读参考输出(每次运行可能不同,size 常 < 10000):
* 期望条目数: 10000 | HashMap 实际 size: 9876 (若小于期望值,说明并发 put 已丢数据)
*/
}
/**
* 经典误用:以为 ConcurrentHashMap 线程安全,check-then-act 就也安全。
* 两个线程可能同时看到 get == null,各自执行昂贵构造并 put。
*/
static int demoGetThenPutBadPattern() throws InterruptedException {
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
AtomicInteger creationCount = new AtomicInteger();
ExecutorService pool = Executors.newFixedThreadPool(THREAD_COUNT);
CountDownLatch startGate = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(THREAD_COUNT);
for (int i = 0; i < THREAD_COUNT; i++) {
pool.execute(() -> {
try {
startGate.await();
if (map.get(CACHE_KEY) == null) {
Thread.sleep(2); // 放大竞态窗口
map.put(CACHE_KEY, createValue(creationCount));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
done.countDown();
}
});
}
startGate.countDown();
done.await();
pool.shutdown();
return creationCount.get();
}
/** 容器提供的原子复合 API:同一 key 只构造一次。 */
static int demoComputeIfAbsent() throws InterruptedException {
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
AtomicInteger creationCount = new AtomicInteger();
ExecutorService pool = Executors.newFixedThreadPool(THREAD_COUNT);
CountDownLatch startGate = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(THREAD_COUNT);
for (int i = 0; i < THREAD_COUNT; i++) {
pool.execute(() -> {
try {
startGate.await();
map.computeIfAbsent(CACHE_KEY, k -> {
try {
Thread.sleep(2);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return createValue(creationCount);
});
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
done.countDown();
}
});
}
startGate.countDown();
done.await();
pool.shutdown();
return creationCount.get();
}
/**
* synchronizedMap 用同一把锁包住整个 map,高并发下吞吐低于 ConcurrentHashMap。
* 本实验只打印两种 map 完成批量 put 的耗时量级差异(非严格压测)。
*/
static void demoSynchronizedMapContention() throws InterruptedException {
int ops = 20_000;
Map<String, Integer> syncMap = Collections.synchronizedMap(new HashMap<>());
Map<String, Integer> concurrentMap = new ConcurrentHashMap<>();
long syncMs = timedBulkPut(syncMap, ops, "sync-map");
long concurrentMs = timedBulkPut(concurrentMap, ops, "chm");
System.out.println("synchronizedMap 耗时: " + syncMs + "ms | ConcurrentHashMap 耗时: "
+ concurrentMs + "ms (并发读写下 CHM 通常更快)");
}
static long timedBulkPut(Map<String, Integer> map, int ops, String prefix)
throws InterruptedException {
int threads = 8;
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch latch = new CountDownLatch(threads);
long start = System.currentTimeMillis();
for (int t = 0; t < threads; t++) {
final int threadId = t;
pool.execute(() -> {
for (int i = 0; i < ops / threads; i++) {
map.put(prefix + "-" + threadId + "-" + i, i);
}
latch.countDown();
});
}
latch.await();
pool.shutdown();
return System.currentTimeMillis() - start;
}
static String createValue(AtomicInteger creationCount) {
creationCount.incrementAndGet();
return "value-" + creationCount.get();
}
}