ConcurrentHashMap案例

本文目录导读：

1. 基础使用案例
2. 计数器与统计
3. 用户会话管理
4. 复杂数据处理
5. 分布式锁的实现
6. 关键知识点

我来为你提供几个ConcurrentHashMap的实际使用案例,从基础到高级场景。

基础使用案例

缓存系统实现

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
public class CacheExample {
    private final ConcurrentHashMap<String, CacheValue> cache = new ConcurrentHashMap<>();
    private volatile boolean running = true;
    // 缓存值包装类，包含过期时间
    private static class CacheValue {
        private final Object value;
        private final long expireTime;
        public CacheValue(Object value, long expireTime) {
            this.value = value;
            this.expireTime = System.currentTimeMillis() + expireTime;
        }
        public boolean isExpired() {
            return System.currentTimeMillis() > expireTime;
        }
        public Object getValue() {
            return value;
        }
    }
    // 添加缓存（带过期时间）
    public void put(String key, Object value, long expireTime) {
        cache.put(key, new CacheValue(value, expireTime));
        // 随机清理过期缓存，避免集中清理
        if (Math.random() < 0.1) {
            cleanExpired();
        }
    }
    // 获取缓存
    public Object get(String key) {
        CacheValue cacheValue = cache.get(key);
        if (cacheValue != null) {
            if (cacheValue.isExpired()) {
                cache.remove(key);
                return null;
            }
            return cacheValue.getValue();
        }
        return null;
    }
    // 清理过期缓存
    public void cleanExpired() {
        long start = System.currentTimeMillis();
        cache.forEach((key, value) -> {
            if (value.isExpired()) {
                cache.remove(key);
            }
        });
        System.out.println("清理缓存耗时: " + (System.currentTimeMillis() - start) + "ms");
    }
    // 启动清理线程
    public void startCleanerThread() {
        new Thread(() -> {
            while (running) {
                try {
                    TimeUnit.SECONDS.sleep(30);
                    cleanExpired();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
    public void shutdown() {
        running = false;
    }
    public static void main(String[] args) throws InterruptedException {
        CacheExample cache = new CacheExample();
        // 启动清理线程
        cache.startCleanerThread();
        // 模拟并发写入
        for (int i = 0; i < 5; i++) {
            final int index = i;
            new Thread(() -> {
                for (int j = 0; j < 10; j++) {
                    String key = "key-" + index + "-" + j;
                    String value = "value-" + index + "-" + j;
                    cache.put(key, value, 5000); // 5秒过期
                    System.out.println("写入: " + key);
                }
            }).start();
        }
        Thread.sleep(1000);
        // 模拟读取
        System.out.println("获取值: " + cache.get("key-0-1"));
        Thread.sleep(6000);
        // 查看过期清理情况
        System.out.println("过期后的值: " + cache.get("key-0-1"));
        cache.shutdown();
    }
}

计数器与统计

多线程计数器

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
public class CounterExample {
    private final ConcurrentHashMap<String, LongAdder> counters = new ConcurrentHashMap<>();
    // 增加计数
    public void increment(String key) {
        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;
    }
    // 获取所有计数
    public void printAllCounts() {
        counters.forEach((key, adder) -> {
            System.out.println(key + ": " + adder.sum());
        });
    }
    public static void main(String[] args) throws InterruptedException {
        CounterExample counter = new CounterExample();
        // 模拟多线程并发计数
        Thread[] threads = new Thread[10];
        for (int i = 0; i < threads.length; i++) {
            final String category = "category-" + (i % 3);
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    counter.increment(category);
                }
            });
            threads[i].start();
        }
        // 等待所有线程完成
        for (Thread thread : threads) {
            thread.join();
        }
        // 输出结果
        counter.printAllCounts();
        System.out.println("总计数: " + 
            (counter.getCount("category-0") + 
             counter.getCount("category-1") + 
             counter.getCount("category-2")));
    }
}

用户会话管理

在线用户会话管理

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
public class SessionManager {
    private static final ConcurrentHashMap<String, UserSession> sessions = new ConcurrentHashMap<>();
    // 用户会话类
    private static class UserSession {
        private final String userId;
        private final String username;
        private volatile long lastAccessTime;
        private volatile boolean isOnline;
        public UserSession(String userId, String username) {
            this.userId = userId;
            this.username = username;
            this.lastAccessTime = System.currentTimeMillis();
            this.isOnline = true;
        }
        public void updateLastAccess() {
            this.lastAccessTime = System.currentTimeMillis();
        }
        public long getLastAccessTime() {
            return lastAccessTime;
        }
        public String getUserId() {
            return userId;
        }
        public String getUsername() {
            return username;
        }
        public void logout() {
            this.isOnline = false;
        }
        public boolean isOnline() {
            return isOnline;
        }
    }
    // 用户登录
    public void login(String userId, String username) {
        UserSession session = new UserSession(userId, username);
        sessions.put(userId, session);
        System.out.println("用户登录: " + username);
    }
    // 用户登出
    public void logout(String userId) {
        UserSession session = sessions.get(userId);
        if (session != null) {
            session.logout();
            sessions.remove(userId);
            System.out.println("用户登出: " + session.getUsername());
        }
    }
    // 更新访问时间（模拟用户活动）
    public void updateUserActivity(String userId) {
        UserSession session = sessions.get(userId);
        if (session != null) {
            session.updateLastAccess();
        }
    }
    // 获取在线用户数量
    public int getOnlineUserCount() {
        return sessions.size();
    }
    // 检查用户是否在线
    public boolean isUserOnline(String userId) {
        return sessions.containsKey(userId);
    }
    // 获取所有在线用户信息
    public void printOnlineUsers() {
        sessions.forEach((userId, session) -> {
            System.out.println("在线用户: " + session.getUsername() + 
                             " (ID: " + userId + ")");
        });
    }
    // 清理超时会话的线程
    public void startSessionCleaner() {
        new Thread(() -> {
            while (true) {
                try {
                    TimeUnit.MINUTES.sleep(1);
                    long currentTime = System.currentTimeMillis();
                    sessions.forEach((userId, session) -> {
                        // 如果30分钟没有活动，则自动登出
                        if (currentTime - session.getLastAccessTime() > 
                            TimeUnit.MINUTES.toMillis(30)) {
                            System.out.println("自动登出: " + session.getUsername());
                            sessions.remove(userId);
                        }
                    });
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
    public static void main(String[] args) throws InterruptedException {
        SessionManager manager = new SessionManager();
        manager.startSessionCleaner();
        // 模拟用户登录
        manager.login("user-001", "张三");
        manager.login("user-002", "李四");
        manager.login("user-003", "王五");
        // 模拟用户活动
        manager.updateUserActivity("user-001");
        System.out.println("\n当前在线用户数: " + manager.getOnlineUserCount());
        manager.printOnlineUsers();
        // 模拟用户登出
        manager.logout("user-002");
        System.out.println("\n登出后在线用户数: " + manager.getOnlineUserCount());
        System.out.println("用户user-002是否在线: " + manager.isUserOnline("user-002"));
        // 等待清理线程运行
        Thread.sleep(2000);
    }
}

复杂数据处理

电商购物车系统

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class ShoppingCartExample {
    // 用户购物车（外层Map：用户 -> 内层Map：商品 -> 数量）
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Integer>> 
        carts = new ConcurrentHashMap<>();
    // 添加商品到购物车
    public void addToCart(String userId, String productId, int quantity) {
        carts.computeIfAbsent(userId, k -> new ConcurrentHashMap<>())
             .merge(productId, quantity, Integer::sum);
        System.out.println("用户" + userId + "添加商品" + productId + 
                         " x" + quantity);
    }
    // 修改商品数量
    public void updateQuantity(String userId, String productId, int newQuantity) {
        ConcurrentHashMap<String, Integer> userCart = carts.get(userId);
        if (userCart != null) {
            if (newQuantity <= 0) {
                userCart.remove(productId);
            } else {
                userCart.put(productId, newQuantity);
            }
        }
    }
    // 查看购物车
    public void viewCart(String userId) {
        ConcurrentHashMap<String, Integer> userCart = carts.get(userId);
        if (userCart == null || userCart.isEmpty()) {
            System.out.println("用户" + userId + "的购物车是空的");
        } else {
            System.out.println("用户" + userId + "的购物车：");
            userCart.forEach((productId, quantity) -> {
                System.out.println("  商品" + productId + ": " + quantity + "个");
            });
        }
    }
    // 计算购物车总商品数
    public int getTotalItems(String userId) {
        ConcurrentHashMap<String, Integer> userCart = carts.get(userId);
        if (userCart == null) return 0;
        return userCart.values().stream().mapToInt(Integer::intValue).sum();
    }
    // 清空购物车
    public void clearCart(String userId) {
        carts.remove(userId);
        System.out.println("清空用户" + userId + "的购物车");
    }
    // 合并购物车（用于A/B测试或跨设备同步）
    public void mergeCarts(String fromUserId, String toUserId) {
        ConcurrentHashMap<String, Integer> fromCart = carts.get(fromUserId);
        ConcurrentHashMap<String, Integer> toCart = carts.get(toUserId);
        if (fromCart != null && toCart != null) {
            fromCart.forEach((productId, quantity) -> {
                toCart.merge(productId, quantity, Integer::sum);
            });
            carts.remove(fromUserId);
            System.out.println("合并购物车: " + fromUserId + " -> " + toUserId);
        }
    }
    public static void main(String[] args) {
        ShoppingCartExample cartSystem = new ShoppingCartExample();
        // 模拟用户购物操作
        Thread addThread1 = new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                cartSystem.addToCart("user-001", "商品-" + i, i);
            }
        });
        Thread addThread2 = new Thread(() -> {
            for (int i = 1; i <= 3; i++) {
                cartSystem.addToCart("user-001", "商品-" + i, 2);
            }
        });
        Thread addThread3 = new Thread(() -> {
            for (int i = 1; i <= 4; i++) {
                cartSystem.addToCart("user-002", "商品-" + i, 1);
            }
        });
        // 启动所有线程
        addThread1.start();
        addThread2.start();
        addThread3.start();
        // 等待线程完成
        try {
            addThread1.join();
            addThread2.join();
            addThread3.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 查看购物车
        cartSystem.viewCart("user-001");
        cartSystem.viewCart("user-002");
        // 修改数量
        cartSystem.updateQuantity("user-001", "商品-1", 1);
        // 计算总商品数
        System.out.println("用户user-001购物车中商品总数: " + 
                         cartSystem.getTotalItems("user-001"));
        // 合并购物车
        cartSystem.mergeCarts("user-002", "user-001");
        // 清空购物车
        cartSystem.clearCart("user-001");
    }
}

分布式锁的实现

简单分布式锁

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.UUID;
import java.util.concurrent.locks.ReentrantLock;
public class DistributedLockExample {
    private static final ConcurrentHashMap<String, LockInfo> locks = 
        new ConcurrentHashMap<>();
    // 锁信息
    private static class LockInfo {
        private final String lockId;
        private final long expireTime;
        public LockInfo(String lockId, long expireTime) {
            this.lockId = lockId;
            this.expireTime = expireTime;
        }
        public String getLockId() {
            return lockId;
        }
        public long getExpireTime() {
            return expireTime;
        }
    }
    // 获取锁
    public boolean acquireLock(String key, long timeout, TimeUnit unit) {
        String lockId = UUID.randomUUID().toString();
        long expireTime = System.currentTimeMillis() + unit.toMillis(timeout);
        LockInfo newLock = new LockInfo(lockId, expireTime);
        LockInfo oldLock = locks.putIfAbsent(key, newLock);
        if (oldLock == null) {
            System.out.println("获取锁成功: " + key);
            return true;
        }
        // 检查锁是否过期
        if (System.currentTimeMillis() > oldLock.getExpireTime()) {
            // 尝试替换过期锁
            boolean updated = locks.replace(key, oldLock, newLock);
            if (updated) {
                System.out.println("获取到过期锁: " + key);
                return true;
            }
        }
        System.out.println("获取锁失败: " + key);
        return false;
    }
    // 释放锁
    public void releaseLock(String key, String lockId) {
        LockInfo lock = locks.get(key);
        if (lock != null && lock.getLockId().equals(lockId)) {
            locks.remove(key, lock);
            System.out.println("释放锁: " + key);
        }
    }
    // 使用锁保护资源
    public static void main(String[] args) {
        DistributedLockExample lockExample = new DistributedLockExample();
        ReentrantLock lock = new ReentrantLock();
        // 模拟多个线程竞争锁
        for (int i = 0; i < 5; i++) {
            final String lockKey = "resource-1";
            new Thread(() -> {
                // 获取锁
                if (lockExample.acquireLock(lockKey, 10, TimeUnit.SECONDS)) {
                    try {
                        System.out.println(Thread.currentThread().getName() + 
                                         " 正在访问共享资源");
                        Thread.sleep(1000); // 模拟资源访问
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } finally {
                        lockExample.releaseLock(lockKey, 
                            lockExample.locks.get(lockKey).getLockId());
                    }
                }
            }).start();
        }
    }
}

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class BestPractices {
    private final ConcurrentMap<String, String> map = new ConcurrentHashMap<>();
    // 1. 使用 putIfAbsent 实现原子操作
    public String getOrAdd(String key, String defaultValue) {
        return map.putIfAbsent(key, defaultValue);
    }
    // 2. 使用 computeIfAbsent 避免竞态条件
    public String computeValue(String key) {
        return map.computeIfAbsent(key, k -> expensiveComputation(k));
    }
    // 3. 使用 compute 进行复杂的原子更新
    public void updateValue(String key) {
        map.compute(key, (k, v) -> 
            v == null ? "default" : v + "-updated");
    }
    // 4. 使用 merge 进行合并操作
    public void mergeValues(String key, String newValue) {
        map.merge(key, newValue, (oldV, newV) -> oldV + "," + newV);
    }
    // 5. 避免使用 synchronized 保护单独操作
    public void avoidRaceCondition(String key, String value) {
        // 错误方式
        /*
        synchronized (map) {
            if (!map.containsKey(key)) {
                map.put(key, value);
            }
        }
        */
        // 正确方式
        map.putIfAbsent(key, value);
    }
    private String expensiveComputation(String key) {
        // 模拟耗时操作
        return "computed-" + key;
    }
}

关键知识点

1. 线程安全性：ConcurrentHashMap 提供高并发下的线程安全操作
2. 原子操作：支持 putIfAbsent、compute、merge 等原子方法
3. 性能优化：分段锁设计，在 JDK8+ 中使用 CAS + synchronized 优化大桶
4. 弱一致性：迭代器是弱一致性的，不保证实时反映最新状态
5. 不适用场景：不建议用于需要强一致性的场景，或频繁的包含检查

这些案例涵盖了 ConcurrentHashMap 最常见的应用场景,你可以根据实际需求选择合适的用法。