AutoMQ Kafka中AsyncObjectLRUCache的并发修改问题分析与解决方案

2025-06-06 06:34:17作者：秋泉律Samson

问题背景

在AutoMQ Kafka项目中，AsyncObjectLRUCache作为LRU缓存实现，用于管理流数据对象的缓存。近期在生产环境中发现了一个ConcurrentModificationException异常，该异常发生在缓存操作过程中，可能导致系统不稳定甚至崩溃。

问题现象

异常堆栈显示，当系统执行以下操作链时会出现问题：

加载流数据对象信息
尝试将对象放入缓存
计算缓存总大小时触发异常

具体表现为LinkedHashMap在迭代过程中被并发修改，抛出ConcurrentModificationException。这种异常通常发生在集合被一个线程迭代的同时，另一个线程修改了集合结构。

根本原因分析

经过深入分析，发现问题根源在于AsyncObjectLRUCache的设计存在以下缺陷：

递归调用风险：在计算缓存总大小(totalSize)的方法中，会触发evict操作，而evict操作又会调用totalSize，形成潜在的递归调用链。
异步回调问题：当调用put方法添加新对象时，会异步获取对象大小(size)，并在获取完成后再次触发evict操作。这个异步回调可能在计算totalSize的过程中执行，导致集合被并发修改。
同步锁粒度不足：虽然方法都标记为synchronized，但由于异步回调的存在，锁无法保护跨线程的操作序列。

技术细节

问题的核心在于以下代码逻辑：

public synchronized void put(K key, V value) {
    evict();  // 第一次evict
    super.put(key, value);
    value.size().thenAccept(v -> evict());  // 异步回调中再次evict
}

private synchronized int totalSize() {
    return cacheEntrySet.stream().mapToInt(entry -> {
        int size = entry.getValue().size().get();  // 同步等待size
        if (随机条件) {
            evict();  // 在迭代过程中可能触发evict
        }
        return size;
    }).sum();
}

当多个线程同时操作缓存时，可能出现以下危险序列：

线程A开始计算totalSize，迭代缓存条目
线程B完成size计算，触发异步回调执行evict
线程A的迭代过程被线程B的修改中断，抛出异常

解决方案

针对这个问题，我们提出以下改进方案：

消除递归调用：重构totalSize方法，确保它不会触发任何可能修改缓存结构的操作。
异步处理优化：将异步回调中的evict操作改为提交到有序任务队列，避免与当前操作冲突。
状态标记：在计算totalSize时设置标记，阻止并发evict操作。

改进后的代码结构如下：

private volatile boolean calculatingSize = false;

public synchronized void put(K key, V value) {
    evict();
    super.put(key, value);
    value.size().thenAccept(v -> {
        if (!calculatingSize) {
            evict();
        }
    });
}

private synchronized int totalSize() {
    calculatingSize = true;
    try {
        return cacheEntrySet.stream()
            .mapToInt(entry -> entry.getValue().size().get())
            .sum();
    } finally {
        calculatingSize = false;
    }
}