前言：在我工作的第四年，负责维护一个核心的商品库存服务。这个服务有一个HashMap<String, Integer> stockMap在内存中缓存所有商品的库存。最初，整个stockMap被一个巨大的synchronized锁保护着。在低并发下相安无事，直到一次大促，这个锁成了整个系统的性能“绞索”，接口RT（响应时间）从10ms飙升到2秒以上。这次痛苦的经历，逼着我完成了一次从“锁粗化”到“锁细化”再到“分段锁”的优化之旅。

一、 问题起源：一把大锁锁全局

先看看最初的问题代码（简化版）：

/**
 * 最初的库存服务（问题版本）
 * 使用一个全局锁保证线程安全
 */
@Service
public class NaiveInventoryService {
    
    // 内存缓存：商品SKU编号 -> 库存数量
    private final Map<String, Integer> stockMap = new HashMap<>();
    
    // 全局锁对象
    private final Object globalLock = new Object();
    
    /**
     * 查询库存
     */
    public Integer getStock(String sku) {
        synchronized (globalLock) { // 查询也要加锁？！是的，为了保证可见性，防止脏读
            return stockMap.getOrDefault(sku, 0);
        }
    }
    
    /**
     * 扣减库存
     */
    public boolean deductStock(String sku, Integer quantity) {
        synchronized (globalLock) {
            Integer currentStock = stockMap.getOrDefault(sku, 0);
            if (currentStock < quantity) {
                return false; // 库存不足
            }
            stockMap.put(sku, currentStock - quantity);
            return true;
        }
    }
    
    /**
     * 增加库存（后台管理用）
     */
    public void addStock(String sku, Integer quantity) {
        synchronized (globalLock) {
            stockMap.put(sku, stockMap.getOrDefault(sku, 0) + quantity);
        }
    }
    
    /**
     * 批量获取多个SKU的库存（订单确认页用）
     */
    public Map<String, Integer> getBatchStock(List<String> skuList) {
        synchronized (globalLock) { // 这个方法会成为灾难！
            Map<String, Integer> result = new HashMap<>();
            for (String sku : skuList) {
                result.put(sku, stockMap.getOrDefault(sku, 0));
            }
            return result;
        }
    }
}

问题分析：

1. 锁粒度太粗：无论操作哪个SKU，无论读写，都用同一把锁。
2. 性能瓶颈：假设有1000个商品，100个并发请求。其中10个请求在修改商品A的库存，另外90个请求在读取商品B、C、D...的库存。这90个读取请求会被那10个写请求完全阻塞，尽管它们操作的是不同的数据！
3. 批量查询是灾难：getBatchStock方法需要遍历多个SKU，持有锁的时间较长，这会阻塞所有其他操作。

用JVisualVM监控线程状态，可以看到大量线程处于BLOCKED状态，证明锁竞争异常激烈。

二、 第一次优化：锁细化（Lock Striping）的尝试

最直观的优化思路是：不同的SKU用不同的锁。这样操作商品A就不会影响操作商品B。

但这里有2个方案：

方案A：为每个SKU创建一个锁对象（不现实）

private final Map<String, Object> skuLocks = new ConcurrentHashMap<>();

public boolean deductStock(String sku, Integer quantity) {
    Object skuLock = skuLocks.computeIfAbsent(sku, k -> new Object());
    synchronized (skuLock) {
        // ... 操作stockMap
    }
}

问题：如果有100万个SKU，就会创建100万个锁对象，内存消耗巨大，而且大多数锁很少被使用。

方案B：分段锁（Segment Lock）
这是更成熟的方案。我们创建固定数量的锁（比如16个），根据SKU的hash值决定使用哪把锁。这样锁的数量是可控的。

/**
 * 第一版优化：基于分段锁的库存服务
 */
@Service
public class SegmentLockInventoryService {
    
    private final Map<String, Integer> stockMap = new HashMap<>();
    
    // 分段锁的数量，一般是2的n次方，方便取模运算
    private static final int SEGMENT_COUNT = 16;
    
    // 分段锁数组
    private final Object[] segmentLocks = new Object[SEGMENT_COUNT];
    
    public SegmentLockInventoryService() {
        // 初始化锁对象
        for (int i = 0; i < SEGMENT_COUNT; i++) {
            segmentLocks[i] = new Object();
        }
    }
    
    /**
     * 根据SKU计算应该使用哪个分段锁
     */
    private Object getLockForSku(String sku) {
        // 使用hashCode并取绝对值，然后对分段数取模
        int segmentIndex = Math.abs(sku.hashCode()) % SEGMENT_COUNT;
        return segmentLocks[segmentIndex];
    }
    
    public Integer getStock(String sku) {
        Object lock = getLockForSku(sku);
        synchronized (lock) {
            return stockMap.getOrDefault(sku, 0);
        }
    }
    
    public boolean deductStock(String sku, Integer quantity) {
        Object lock = getLockForSku(sku);
        synchronized (lock) {
            Integer currentStock = stockMap.getOrDefault(sku, 0);
            if (currentStock < quantity) {
                return false;
            }
            stockMap.put(sku, currentStock - quantity);
            return true;
        }
    }
    
    // ... 其他方法类似
}

优化效果：

• 锁竞争从全局级别降低到了分段级别。理想情况下，性能应该提升接近SEGMENT_COUNT倍（16倍）。
• 内存消耗固定，只有16个锁对象。

但带来了新问题：

1. 批量查询更复杂了：getBatchStock需要获取多个锁，而且必须按固定顺序获取，否则可能产生死锁。
2. 数据一致性视图：批量查询时，如果依次获取每个SKU的锁，在查询过程中，其他SKU的库存可能已经改变，无法获得一致性的快照。

三、 第二次优化：读写锁（ReadWriteLock）的引入

观察我们的业务：读多写少。查询库存的QPS远高于扣减库存的QPS。synchronized是互斥锁，不区分读写。我们可以使用ReadWriteLock来优化。

/**
 * 第二版优化：分段锁 + 读写锁
 */
@Service
public class ReadWriteSegmentInventoryService {
    
    private final Map<String, Integer> stockMap = new HashMap<>();
    private static final int SEGMENT_COUNT = 16;
    
    // 分段锁数组，现在是ReadWriteLock
    private final ReadWriteLock[] segmentLocks = new ReadWriteLock[SEGMENT_COUNT];
    
    public ReadWriteSegmentInventoryService() {
        for (int i = 0; i < SEGMENT_COUNT; i++) {
            // 使用公平锁还是非公平锁？根据场景选择，这里用非公平锁（默认）追求更高吞吐量
            segmentLocks[i] = new ReentrantReadWriteLock();
        }
    }
    
    private ReadWriteLock getLockForSku(String sku) {
        int segmentIndex = Math.abs(sku.hashCode()) % SEGMENT_COUNT;
        return segmentLocks[segmentIndex];
    }
    
    /**
     * 读操作使用读锁，可以并发执行
     */
    public Integer getStock(String sku) {
        ReadWriteLock lock = getLockForSku(sku);
        lock.readLock().lock();
        try {
            return stockMap.getOrDefault(sku, 0);
        } finally {
            lock.readLock().unlock();
        }
    }
    
    /**
     * 写操作使用写锁，互斥执行
     */
    public boolean deductStock(String sku, Integer quantity) {
        ReadWriteLock lock = getLockForSku(sku);
        lock.writeLock().lock();
        try {
            Integer currentStock = stockMap.getOrDefault(sku, 0);
            if (currentStock < quantity) {
                return false;
            }
            stockMap.put(sku, currentStock - quantity);
            return true;
        } finally {
            lock.writeLock().unlock();
        }
    }
}

优化效果：

• 对同一个分段的多个读操作可以完全并发，进一步提升了读性能。
• 写操作仍然互斥，保证数据安全。

但还是没解决批量查询的问题。

四、 终极方案：使用ConcurrentHashMap

在苦苦思考如何实现安全的批量查询时，我猛然意识到：我TM不是在重新发明ConcurrentHashMap吗？

ConcurrentHashMap在JDK中的实现，正是分段锁思想的生产级实现！而且在JDK 8之后，还优化为了synchronized + CAS的实现，性能更好。

豁然开朗后的重构：

/**
 * 终极优化：直接使用ConcurrentHashMap，让JDK大神来帮我们解决并发问题
 */
@Service
public class ConcurrentHashMapInventoryService {
    
    // ConcurrentHashMap本身就是线程安全的，不需要额外的锁！
    private final ConcurrentMap<String, Integer> stockMap = new ConcurrentHashMap<>();
    
    public Integer getStock(String sku) {
        // 简单到令人发指！
        return stockMap.getOrDefault(sku, 0);
    }
    
    public boolean deductStock(String sku, Integer quantity) {
        // 使用CAS乐观锁机制，避免不必要的阻塞
        while (true) {
            Integer currentStock = stockMap.get(sku);
            if (currentStock == null) {
                // 商品不存在，可以初始化或者返回失败
                return false;
            }
            if (currentStock < quantity) {
                return false;
            }
            
            // 使用CAS更新：如果当前值还是currentStock，就更新为currentStock - quantity
            if (stockMap.replace(sku, currentStock, currentStock - quantity)) {
                return true; // 更新成功
            }
            // 如果CAS失败，说明有其他线程修改了库存，重试！
        }
    }
    
    /**
     * 批量查询：ConcurrentHashMap的get操作是无锁的，性能极高
     * 但注意：这仍然不是"一致性快照"，只是瞬间的状态
     */
    public Map<String, Integer> getBatchStock(List<String> skuList) {
        Map<String, Integer> result = new HashMap<>();
        for (String sku : skuList) {
            result.put(sku, stockMap.getOrDefault(sku, 0));
        }
        return result;
    }
    
    /**
     * 更优雅的扣减库存方式：使用compute方法
     */
    public boolean deductStockV2(String sku, Integer quantity) {
        return stockMap.compute(sku, (key, currentStock) -> {
            if (currentStock == null) return 0; // 或者抛异常
            if (currentStock < quantity) {
                throw new RuntimeException("库存不足"); // 返回null会删除key，所以用异常
            }
            return currentStock - quantity;
        }) != null;
    }
}

为什么这是终极方案？

1. 站在巨人的肩膀上：ConcurrentHashMap是JDK专家精心优化的线程安全容器，比自己实现的分段锁更可靠、性能更好。
2. 更细的锁粒度：在JDK 8中，ConcurrentHashMap的锁粒度是每个桶（bucket），比固定分段更细。
3. 使用CAS无锁操作：对于大多数读操作和部分写操作，使用CAS避免加锁，性能极高。
4. API丰富：提供了compute、merge等原子操作方法，让业务逻辑更简洁。

五、 针对特殊场景的进一步优化

但我们的业务还有特殊要求：有时候需要真正的一致性快照（比如生成库存报表时，希望所有SKU的数据是同一时刻的）。

对于这种真正需要全局锁的场景，我们不应该逃避，而是应该控制它的影响范围：

/**
 * 混合方案：平时用ConcurrentHashMap的无锁高性能，特殊场景才用全局锁
 */
@Service
public class HybridInventoryService {
    
    private final ConcurrentMap<String, Integer> stockMap = new ConcurrentHashMap<>();
    
    // 用于全局操作的锁，尽量少用！
    private final ReentrantLock globalLock = new ReentrantLock();
    
    // 平时的单个操作，用无锁方式
    public Integer getStock(String sku) {
        return stockMap.getOrDefault(sku, 0);
    }
    
    public boolean deductStock(String sku, Integer quantity) {
        // 使用之前的CAS方案
        // ... 
    }
    
    /**
     * 真正的一致性快照批量查询（用于报表生成）
     * 慎用！会阻塞所有写操作！
     */
    public Map<String, Integer> getConsistentBatchStock(List<String> skuList) {
        globalLock.lock();  // 获取全局锁
        try {
            // 这里可以放心地遍历，因为所有写操作都被阻塞了
            Map<String, Integer> result = new HashMap<>();
            for (String sku : skuList) {
                result.put(sku, stockMap.getOrDefault(sku, 0));
            }
            return result;
        } finally {
            globalLock.unlock();
        }
    }
    
    /**
     * 写操作也要支持全局锁
     */
    public boolean deductStockWithGlobalLock(String sku, Integer quantity) {
        globalLock.lock();
        try {
            return doDeductStock(sku, quantity); // 实际的扣减逻辑
        } finally {
            globalLock.unlock();
        }
    }
}

关键洞察：不要追求绝对的"无锁"，而是要根据业务场景选择最合适的锁粒度。

六、 锁优化总结：一个清晰的演进路径

通过这次优化之旅，我总结出了锁优化的清晰路径：

1. 粗粒度锁（发现问题）：简单粗暴，适用于原型阶段或低并发场景。
2. 锁细化（思考方案）：意识到不同数据应该用不同的锁。
3. 分段锁（手动实现）：平衡锁粒度和内存开销的实用方案。
4. 读写锁（进一步优化）：针对读多写少场景的特化优化。
5. 使用并发容器（最佳实践）：ConcurrentHashMap、ConcurrentSkipListMap等是生产级解决方案。
6. CAS无锁编程（高级技巧）：在特定场景下用原子变量避免锁。
7. 混合方案（实事求是）：根据业务特点，在需要时谨慎使用全局锁。

七、 更深层次的思考

这次优化让我明白了几点：

1. 不要过早优化：如果我们的系统永远只有10QPS，最初的全局锁方案是最简单的。
2. 测量，不要猜测：用jstack、JVisualVM等工具证明锁竞争的存在，用压测数据验证优化效果。
3. 理解JDK源码：ConcurrentHashMap的源码是最好的学习材料，理解了它的实现，就理解了现代Java并发编程的精髓。
4. 业务决定技术：锁优化的目标不是追求技术上的"完美"，而是满足业务场景下的性能要求。

从一把粗笨的全局锁，到精致的分段锁，再到直接使用ConcurrentHashMap，这个过程让我真正理解了Java并发编程的艺术：在保证线程安全的前提下，找到性能与复杂度的最佳平衡点。