本文目录导读：
我来为你详细讲解ConcurrentHashMap的实战案例，涵盖常见使用场景和完整示例代码。
基础使用案例
1 缓存管理器
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
* 基于ConcurrentHashMap的缓存管理器
*/
public class CacheManager<K, V> {
// 缓存存储
private final ConcurrentHashMap<K, CacheEntry<V>> cacheMap = new ConcurrentHashMap<>();
// 缓存命中统计
private final AtomicLong hitCount = new AtomicLong(0);
private final AtomicLong missCount = new AtomicLong(0);
// 默认过期时间
private final long defaultExpireTime;
/**
* 缓存条目
*/
private static class CacheEntry<V> {
private final V value;
private final long expireTime;
public CacheEntry(V value, long expireTime) {
this.value = value;
this.expireTime = expireTime;
}
public boolean isExpired() {
return System.currentTimeMillis() > expireTime;
}
}
public CacheManager(long defaultExpireTime) {
this.defaultExpireTime = defaultExpireTime;
}
/**
* 添加缓存
*/
public void put(K key, V value) {
put(key, value, defaultExpireTime);
}
/**
* 添加缓存并指定过期时间
*/
public void put(K key, V value, long expireTime) {
CacheEntry<V> entry = new CacheEntry<>(value, System.currentTimeMillis() + expireTime);
cacheMap.put(key, entry);
}
/**
* 获取缓存（自动处理过期）
*/
public V get(K key) {
CacheEntry<V> entry = cacheMap.get(key);
if (entry == null) {
missCount.incrementAndGet();
return null;
}
// 检查是否过期
if (entry.isExpired()) {
cacheMap.remove(key);
missCount.incrementAndGet();
return null;
}
hitCount.incrementAndGet();
return entry.value;
}
/**
* 如果不存在则添加，返回现有值
*/
public V putIfAbsent(K key, V value) {
CacheEntry<V> newEntry = new CacheEntry<>(value, System.currentTimeMillis() + defaultExpireTime);
CacheEntry<V> existing = cacheMap.putIfAbsent(key, newEntry);
if (existing == null) {
return value;
}
return existing.value;
}
/**
* 删除缓存
*/
public void remove(K key) {
cacheMap.remove(key);
}
/**
* 清空缓存
*/
public void clear() {
cacheMap.clear();
}
/**
* 获取缓存大小
*/
public int size() {
return cacheMap.size();
}
/**
* 获取命中率
*/
public double getHitRate() {
long hits = hitCount.get();
long misses = missCount.get();
long total = hits + misses;
return total == 0 ? 0 : (double) hits / total;
}
// 测试示例
public static void main(String[] args) throws InterruptedException {
// 创建缓存管理器，默认过期时间5秒
CacheManager<String, String> cache = new CacheManager<>(5000);
// 存储数据
cache.put("user:1", "张三");
cache.put("user:2", "李四");
cache.put("temp", "临时数据", 2000); // 2秒后过期
// 获取数据
System.out.println("user:1 = " + cache.get("user:1"));
System.out.println("user:2 = " + cache.get("user:2"));
// 等待2秒后获取临时数据
Thread.sleep(2000);
System.out.println("temp (过期后) = " + cache.get("temp"));
// 使用putIfAbsent
cache.putIfAbsent("user:1", "王五");
System.out.println("user:1 (putIfAbsent) = " + cache.get("user:1"));
// 统计信息
System.out.println("缓存大小: " + cache.size());
System.out.println("命中率: " + cache.getHitRate());
}
}
2 高并发计数器
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
/**
* 高并发计数器
*/
public class ConcurrentCounter {
// 使用LongAdder提高并发性能
private final ConcurrentHashMap<String, LongAdder> counters = new ConcurrentHashMap<>();
// 或者使用AtomicLong
private final ConcurrentHashMap<String, AtomicLong> atomicCounters = new ConcurrentHashMap<>();
/**
* 增加计数（使用LongAdder）
*/
public void increment(String key) {
// computeIfAbsent确保线程安全
counters.computeIfAbsent(key, k -> new LongAdder()).increment();
}
/**
* 增加指定数量
*/
public void add(String key, long value) {
counters.computeIfAbsent(key, k -> new LongAdder()).add(value);
}
/**
* 获取计数
*/
public long getCount(String key) {
LongAdder adder = counters.get(key);
return adder != null ? adder.sum() : 0;
}
/**
* 使用AtomicLong实现
*/
public void incrementWithAtomic(String key) {
atomicCounters.computeIfAbsent(key, k -> new AtomicLong(0)).incrementAndGet();
}
/**
* 复杂操作 - 原子更新
*/
public void updateIfMatches(String key, long expectedValue, long newValue) {
atomicCounters.compute(key, (k, current) -> {
if (current == null) {
current = new AtomicLong(0);
}
// 原子地获取和更新
current.compareAndSet(expectedValue, newValue);
return current;
});
}
/**
* 批量操作
*/
public void batchIncrement(String... keys) {
for (String key : keys) {
increment(key);
}
}
public static void main(String[] args) throws InterruptedException {
ConcurrentCounter counter = new ConcurrentCounter();
// 多线程并发计数
int threadCount = 100;
int incrementsPerThread = 10000;
Thread[] threads = new Thread[threadCount];
for (int i = 0; i < threadCount; i++) {
final int threadId = i;
threads[i] = new Thread(() -> {
for (int j = 0; j < incrementsPerThread; j++) {
counter.increment("visits");
counter.increment("thread:" + (threadId % 10));
}
});
threads[i].start();
}
// 等待所有线程完成
for (Thread thread : threads) {
thread.join();
}
// 验证结果
System.out.println("总访问次数: " + counter.getCount("visits"));
System.out.println("预期总次数: " + (threadCount * incrementsPerThread));
System.out.println("Thread 0 计数: " + counter.getCount("thread:0"));
}
}
高级应用场景
1 分布式锁管理器
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
/**
* 简单的分布式锁实现
*/
public class DistributedLockManager {
// 存储锁信息
private final ConcurrentHashMap<String, LockInfo> locks = new ConcurrentHashMap<>();
private static class LockInfo {
private final String ownerId; // 锁持有者ID
private final long acquireTime; // 获取时间
private final long expiryDuration; // 过期时长
private final ReentrantLock lock; // 可重入锁
public LockInfo(String ownerId, long acquireTime, long expiryDuration) {
this.ownerId = ownerId;
this.acquireTime = acquireTime;
this.expiryDuration = expiryDuration;
this.lock = new ReentrantLock(true);
}
public boolean isExpired() {
return System.currentTimeMillis() - acquireTime > expiryDuration;
}
}
/**
* 尝试获取锁
*/
public boolean tryAcquire(String lockKey, String ownerId, long timeoutMillis) {
LockInfo newLock = new LockInfo(ownerId, System.currentTimeMillis(), timeoutMillis);
// 尝试获取锁
LockInfo existing = locks.putIfAbsent(lockKey, newLock);
if (existing == null) {
return true; // 成功获取锁
}
// 检查现有锁是否过期
if (existing.isExpired()) {
// 使用remove和putIfAbsent实现原子替换
boolean removed = locks.remove(lockKey, existing);
if (removed) {
LockInfo replaced = locks.putIfAbsent(lockKey, newLock);
return replaced == null || replaced == newLock;
}
}
return false; // 获取锁失败
}
/**
* 尝试获取锁（带重试）
*/
public boolean tryAcquireWithRetry(String lockKey, String ownerId, long timeoutMillis, int retryCount) {
for (int i = 0; i < retryCount; i++) {
if (tryAcquire(lockKey, ownerId, timeoutMillis)) {
return true;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
return false;
}
/**
* 释放锁（只有拥有者才能释放）
*/
public boolean release(String lockKey, String ownerId) {
LockInfo current = locks.get(lockKey);
if (current != null && current.ownerId.equals(ownerId)) {
// 原子操作：只有值匹配时才删除
return locks.remove(lockKey, current);
}
return false;
}
/**
* 检查锁是否存在
*/
public boolean isLocked(String lockKey) {
return locks.containsKey(lockKey);
}
public static void main(String[] args) {
DistributedLockManager lockManager = new DistributedLockManager();
// 模拟分布式环境中的多个节点
String lockKey = "ORDER_LOCK_12345";
// 节点1尝试获取锁
boolean acquired1 = lockManager.tryAcquire(lockKey, "NODE-1", 5000);
System.out.println("Node-1 获取锁: " + acquired1);
// 节点2尝试获取锁（应该失败）
boolean acquired2 = lockManager.tryAcquire(lockKey, "NODE-2", 5000);
System.out.println("Node-2 获取锁: " + acquired2);
// Node-1释放锁
boolean released = lockManager.release(lockKey, "NODE-1");
System.out.println("Node-1 释放锁: " + released);
// Node-2再次尝试（应该成功）
boolean acquired3 = lockManager.tryAcquire(lockKey, "NODE-2", 5000);
System.out.println("Node-2 重新获取锁: " + acquired3);
}
}
2 数据聚合统计器
import java.util.concurrent.ConcurrentHashMap;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
/**
* 高并发数据聚合统计器
*/
public class DataAggregator {
// 存储每个维度的统计数据
private final ConcurrentHashMap<String, ConcurrentHashMap<String, StatisticData>> stats = new ConcurrentHashMap<>();
private static class StatisticData {
private final AtomicLong count = new AtomicLong(0);
private final AtomicLong sum = new AtomicLong(0);
private final AtomicLong min = new AtomicLong(Long.MAX_VALUE);
private final AtomicLong max = new AtomicLong(Long.MIN_VALUE);
public void addValue(long value) {
count.incrementAndGet();
sum.addAndGet(value);
// 原子更新min和max
while (true) {
long currentMin = min.get();
if (value >= currentMin || min.compareAndSet(currentMin, value)) {
break;
}
}
while (true) {
long currentMax = max.get();
if (value <= currentMax || max.compareAndSet(currentMax, value)) {
break;
}
}
}
public double getAverage() {
long c = count.get();
return c == 0 ? 0 : (double) sum.get() / c;
}
}
/**
* 记录数据
*/
public void record(String category, String dimension, long value) {
stats.computeIfAbsent(category, k -> new ConcurrentHashMap<>())
.computeIfAbsent(dimension, k -> new StatisticData())
.addValue(value);
}
/**
* 获取某维度的统计
*/
public Map<String, Object> getStatistic(String category, String dimension) {
ConcurrentHashMap<String, StatisticData> categoryStats = stats.get(category);
if (categoryStats == null) return Collections.emptyMap();
StatisticData data = categoryStats.get(dimension);
if (data == null) return Collections.emptyMap();
Map<String, Object> result = new HashMap<>();
result.put("count", data.count.get());
result.put("sum", data.sum.get());
result.put("min", data.min.get());
result.put("max", data.max.get());
result.put("average", data.getAverage());
return result;
}
/**
* 获取某类别所有维度的汇总
*/
public Map<String, Map<String, Object>> getCategorySummary(String category) {
ConcurrentHashMap<String, StatisticData> categoryStats = stats.get(category);
if (categoryStats == null) return Collections.emptyMap();
return categoryStats.entrySet().stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> {
StatisticData data = entry.getValue();
Map<String, Object> map = new HashMap<>();
map.put("count", data.count.get());
map.put("sum", data.sum.get());
map.put("min", data.min.get());
map.put("max", data.max.get());
map.put("average", data.getAverage());
return map;
}
));
}
public static void main(String[] args) throws InterruptedException {
DataAggregator aggregator = new DataAggregator();
// 模拟多个线程并发写入数据
int threadCount = 50;
Thread[] threads = new Thread[threadCount];
for (int i = 0; i < threadCount; i++) {
final int threadId = i;
threads[i] = new Thread(() -> {
Random random = new Random();
for (int j = 0; j < 1000; j++) {
String category = "category_" + (threadId % 5);
String dimension = "dimension_" + (threadId % 10);
long value = random.nextInt(1000);
aggregator.record(category, dimension, value);
}
});
threads[i].start();
}
// 等待所有线程完成
for (Thread thread : threads) {
thread.join();
}
// 输出统计结果
System.out.println("=== 统计数据 ===");
for (int i = 0; i < 5; i++) {
System.out.println("Category " + i + ":");
Map<String, Map<String, Object>> categoryStats = aggregator.getCategorySummary("category_" + i);
categoryStats.forEach((dimension, stats) -> {
System.out.println(" " + dimension + ": " + stats);
});
}
}
}
最佳实践案例
1 配置中心
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
/**
* 动态配置中心
*/
public class ConfigurationCenter {
private final ConcurrentHashMap<String, ConfigValue> configs = new ConcurrentHashMap<>();
private static class ConfigValue {
private final String value;
private final long version;
private final long timestamp;
private final List<ConfigChangeListener> listeners = new CopyOnWriteArrayList<>();
public ConfigValue(String value, long version, long timestamp) {
this.value = value;
this.version = version;
this.timestamp = timestamp;
}
public void addListener(ConfigChangeListener listener) {
listeners.add(listener);
}
public void notifyListeners(String key) {
listeners.forEach(listener -> listener.onChange(key, value));
}
}
@FunctionalInterface
public interface ConfigChangeListener {
void onChange(String key, String newValue);
}
/**
* 设置配置（带版本管理）
*/
public boolean setConfig(String key, String value, long newVersion) {
ConfigValue existing = configs.get(key);
// 使用compute实现原子操作
ConfigValue[] result = new ConfigValue[1];
configs.compute(key, (k, current) -> {
if (current == null || newVersion > current.version) {
ConfigValue newConfig = new ConfigValue(value, newVersion, System.currentTimeMillis());
// 复制监听器
if (current != null && current.listeners != null) {
current.listeners.forEach(newConfig::addListener);
}
result[0] = newConfig;
return newConfig;
}
result[0] = current;
return current;
});
// 通知监听器
if (result[0] != null && !result[0].value.equals(value)) {
result[0].notifyListeners(key);
return true;
}
return false;
}
/**
* 获取配置值
*/
public String getConfig(String key) {
ConfigValue config = configs.get(key);
return config != null ? config.value : null;
}
/**
* 获取配置值（带默认值）
*/
public String getConfig(String key, String defaultValue) {
String value = getConfig(key);
return value != null ? value : defaultValue;
}
/**
* 添加配置监听器
*/
public void addListener(String key, ConfigChangeListener listener) {
configs.computeIfAbsent(key, k -> new ConfigValue(null, 0, 0))
.addListener(listener);
}
public static void main(String[] args) {
ConfigurationCenter configCenter = new ConfigurationCenter();
// 添加监听器
configCenter.addListener("db.pool.size", (key, value) -> {
System.out.println("配置变更 - " + key + ": " + value);
});
// 设置配置
configCenter.setConfig("db.url", "jdbc:mysql://localhost:3306/db", 1);
configCenter.setConfig("db.pool.size", "10", 1);
// 获取配置
System.out.println("数据库URL: " + configCenter.getConfig("db.url"));
System.out.println("连接池大小: " + configCenter.getConfig("db.pool.size", "5"));
// 更新配置
configCenter.setConfig("db.pool.size", "20", 2);
System.out.println("更新后连接池大小: " + configCenter.getConfig("db.pool.size"));
}
}
2 在线用户管理器
import java.util.concurrent.ConcurrentHashMap;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.stream.Collectors;
/**
* 在线用户管理器
*/
public class OnlineUserManager {
private final ConcurrentHashMap<String, UserSession> activeUsers = new ConcurrentHashMap<>();
private static class UserSession {
private final String userId;
private final String username;
private final String sessionId;
private volatile LocalDateTime lastActiveTime;
private volatile String status;
public UserSession(String userId, String username, String sessionId) {
this.userId = userId;
this.username = username;
this.sessionId = sessionId;
this.lastActiveTime = LocalDateTime.now();
this.status = "ONLINE";
}
public void updateActivity() {
this.lastActiveTime = LocalDateTime.now();
}
public boolean isActive(long timeoutMinutes) {
return lastActiveTime.plusMinutes(timeoutMinutes).isAfter(LocalDateTime.now());
}
}
/**
* 用户上线
*/
public UserSession userLogin(String userId, String username, String sessionId) {
UserSession session = new UserSession(userId, username, sessionId);
UserSession existing = activeUsers.putIfAbsent(userId, session);
return existing != null ? existing : session;
}
/**
* 用户下线
*/
public boolean userLogout(String userId, String sessionId) {
UserSession session = activeUsers.get(userId);
if (session != null && session.sessionId.equals(sessionId)) {
return activeUsers.remove(userId, session);
}
return false;
}
/**
* 更新用户活动状态
*/
public void updateUserActivity(String userId) {
UserSession session = activeUsers.get(userId);
if (session != null) {
session.updateActivity();
}
}
/**
* 获取在线用户数
*/
public int getOnlineCount() {
return activeUsers.size();
}
/**
* 获取所有在线用户
*/
public Map<String, String> getAllOnlineUsers() {
return activeUsers.entrySet().stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> entry.getValue().username
));
}
/**
* 清理过期用户（定时任务调用）
*/
public int cleanupExpiredSessions(long timeoutMinutes) {
int count = 0;
for (Map.Entry<String, UserSession> entry : activeUsers.entrySet()) {
if (!entry.getValue().isActive(timeoutMinutes)) {
if (activeUsers.remove(entry.getKey(), entry.getValue())) {
count++;
}
}
}
return count;
}
public static void main(String[] args) throws InterruptedException {
OnlineUserManager userManager = new OnlineUserManager();
// 模拟用户登录
userManager.userLogin("user1", "张三", "session-1");
userManager.userLogin("user2", "李四", "session-2");
userManager.userLogin("user3", "王五", "session-3");
System.out.println("当前在线用户数: " + userManager.getOnlineCount());
System.out.println("在线用户: " + userManager.getAllOnlineUsers());
// 更新用户活动
userManager.updateUserActivity("user1");
// 模拟用户下线
userManager.userLogout("user2", "session-2");
Thread.sleep(1000);
System.out.println("下线后在线用户数: " + userManager.getOnlineCount());
// 清理过期会话
int cleaned = userManager.cleanupExpiredSessions(1);
System.out.println("清理过期会话: " + cleaned + " 个");
}
}
性能对比案例
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;
/**
* 性能对比测试
*/
public class PerformanceComparison {
private static final int THREAD_COUNT = 100;
private static final int OPERATIONS_PER_THREAD = 100000;
public static void main(String[] args) throws InterruptedException {
// 对比 HashMap（非线程安全）
Map<String, Integer> hashMap = new HashMap<>();
// 对比 ConcurrentHashMap
Map<String, Integer> concurrentMap = new ConcurrentHashMap<>();
System.out.println("=== HashMap 性能测试（线程不安全） ===");
testMap(hashMap, false);
System.out.println("\n=== ConcurrentHashMap 性能测试（线程安全） ===");
testMap(concurrentMap, true);
System.out.println("\n=== ConcurrentHashMap 特殊操作演示 ===");
demonstrateSpecialOperations();
}
private static void testMap(Map<String, Integer> map, boolean threadSafe) throws InterruptedException {
if (threadSafe) {
// 线程安全测试
ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
CountDownLatch latch = new CountDownLatch(THREAD_COUNT);
long startTime = System.currentTimeMillis();
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadId = i;
executor.submit(() -> {
try {
for (int j = 0; j < OPERATIONS_PER_THREAD; j++) {
String key = "key-" + (threadId * 100 + j);
// 使用computeIfAbsent保证原子性
map.computeIfAbsent(key, k -> 0);
// 更新操作
map.compute(key, (k, v) -> (v == null ? 0 : v) + 1);
}
} finally {
latch.countDown();
}
});
}
latch.await();
executor.shutdown();
long endTime = System.currentTimeMillis();
System.out.println("总耗时: " + (endTime - startTime) + "ms");
System.out.println("总操作数: " + THEAD_COUNT * OPERATIONS_PER_THREAD);
System.out.println("Map大小: " + map.size());
} else {
// 非线程安全简单测试
long startTime = System.currentTimeMillis();
for (int i = 0; i < THREAD_COUNT * OPERATIONS_PER_THREAD; i++) {
map.put("key-" + i, i);
Integer value = map.get("key-" + i);
}
long endTime = System.currentTimeMillis();
System.out.println("总耗时: " + (endTime - startTime) + "ms");
System.out.println("Map大小: " + map.size());
}
}
private static void demonstrateSpecialOperations() {
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
// 1. putIfAbsent - 原子性地不超过值不存在时放入
map.putIfAbsent("key1", "value1");
map.putIfAbsent("key1", "value2"); // 不会覆盖
System.out.println("putIfAbsent: " + map.get("key1"));
// 2. compute - 根据现有值计算新值
map.compute("key1", (key, value) -> value + " updated");
System.out.println("compute: " + map.get("key1"));
// 3. merge - 合并操作
map.merge("key2", "initial", (oldValue, newValue) -> oldValue + "+" + newValue);
map.merge("key2", "added", (oldValue, newValue) -> oldValue + "+" + newValue);
System.out.println("merge: " + map.get("key2"));
// 4. forEach - 并行遍历
map.forEach((key, value) -> {
System.out.println("遍历: " + key + " -> " + value);
});
// 5. reduce - 归约操作
int sumLength = map.reduceKeys(2, (key1, key2) -> key1 + key2).length();
System.out.println("所有key长度之和: " + sumLength);
}
}
使用注意事项
1 正确的使用模式
public class ConcurrentHashMapBestPractice {
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
/**
* 正确的使用方式
*/
public void correctUsage() {
// 1. 使用putIfAbsent而不是containsKey+put
Object value = cache.putIfAbsent("key", "value");
// 2. 使用computeIfAbsent处理初始化
Object lazyValue = cache.computeIfAbsent("key", k ->
new Object()
);
// 3. 使用compute进行原子更新
cache.compute("key", (k, v) -> {
if (v == null) return "new";
return v + " updated";
});
// 4. 使用merge合并值
cache.merge("key", 1, (oldValue, newValue) ->
(Integer)oldValue + (Integer)newValue
);
}
/**
* 错误的使用方式（非原子操作）
*/
public void incorrectUsage() {
// 错误：不是原子操作
if (!cache.containsKey("key")) {
cache.put("key", "value");
}
// 错误：get和put不是原子操作
Object value = cache.get("key");
if (value == null) {
cache.put("key", "newValue");
}
}
}
这些案例覆盖了ConcurrentHashMap的主要使用场景，包括：
- 缓存管理 - 最常用的场景
- 高并发计数 - 利用原子类
- 分布式锁 - 基于CAS操作
- 数据聚合 - 统计场景
- 配置中心 - 监听和版本管理
- 在线用户管理 - 状态管理
建议根据实际业务需求选择合适的场景,并注意使用putIfAbsent、compute等原子操作方法来保证并发安全。
