使用Mutex保护协程共享列表仍触发ConcurrentModificationException

我开发了一个TrainService类，持有供多协程共享的Train列表，并在修改该列表的方法中使用Mutex实现互斥锁，期望保证同一时间仅有一个协程能读写该列表。

调用协程代码 #

val contours: List<MatOfPoint> = ArrayList()
val hierarchy = Mat()

Imgproc.findContours(result, contours, hierarchy, Imgproc.RETR_LIST, Imgproc.CHAIN_APPROX_SIMPLE)

Imgproc.cvtColor(result, result, Imgproc.COLOR_GRAY2BGR)

for (contour in contours) {
    if (contour.toArray().size > 100) {
       val mask = Mat.zeros(foreground.size(), CvType.CV_8UC1)
       val white = Scalar.all(255.0)
       Imgproc.drawContours(mask, listOf(contour), -1, white, -1)

       val meanColor = Core.mean(foreground, mask)

       trainService.addTrain(contour, meanColor)

       Imgproc.drawContours(result, listOf(contour), -1, meanColor, FILLED)
    } else {
        Imgproc.drawContours(result, listOf(contour), -1, Scalar(0.0, 0.0, 0.0), FILLED)
    }
}

trainService.removeUnusedTrains()

val resultBitmap =
            Bitmap.createBitmap(result.width(), result.height(), Bitmap.Config.ARGB_8888)
Utils.matToBitmap(result, resultBitmap)

TrainService类代码 #

class TrainService {
    private var trains : MutableList<Train> = emptyList<Train>().toMutableList()

    private val mutex = Mutex()

    suspend fun addTrain(newTrain: MatOfPoint, trainColor: Scalar) = mutex.withLock {
        for(train in trains) {
            when {
                train.isSame(newTrain) -> return true
            }
        }
        val train = Train(newTrain, trainColor)
        trains.add(train)
    }

    suspend fun removeUnusedTrains() = mutex.withLock {
        for(train in trains) {
            if (!train.changed) {
                trains.remove(train)
            } else {
                train.changed = false
            }
        }
        Log.d("TrainService", "Trains now: ${trains.joinToString()}")
    }
}

运行时异常 #

time: 1709989371723
msg: java.util.ConcurrentModificationException
stacktrace: java.util.ConcurrentModificationException
    at java.util.ArrayList$Itr.next(ArrayList.java:860)
    at com.xcerna11.traincontroller.TrainService.removeUnusedTrains(TrainService.kt:25)
    at com.xcerna11.traincontroller.OpenCVDetector$run$2.invokeSuspend(OpenCVDetector.kt:74)
    at com.xcerna11.traincontroller.OpenCVDetector$run$2.invoke(Unknown Source:8)
    at com.xcerna11.traincontroller.OpenCVDetector$run$2.invoke(Unknown Source:4)
    at kotlinx.coroutines.intrinsics.UndispatchedKt.startUndispatchedOrReturn(Undispatched.kt:89)
    at kotlinx.coroutines.CoroutineScopeKt.coroutineScope(CoroutineScope.kt:264)
    at com.xcerna11.traincontroller.OpenCVDetector.run(OpenCVDetector.kt:24)
    at com.xcerna11.traincontroller.TrainAnalyzer$analyze$1$1.invokeSuspend(TrainAnalyzer.kt:20)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
    at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:284)
    at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:85)
    at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
    at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source:1)
    at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
    at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source:1)
    at com.xcerna11.traincontroller.TrainAnalyzer.analyze(TrainAnalyzer.kt:19)
    at androidx.camera.core.ImageAnalysis.lambda$setAnalyzer$2(ImageAnalysis.java:528)
    at androidx.camera.core.ImageAnalysis$$ExternalSyntheticLambda2.analyze(D8$$SyntheticClass:0)
    at androidx.camera.core.ImageAnalysisAbstractAnalyzer.lambda$analyzeImage$0$androidx-camera-core-ImageAnalysisAbstractAnalyzer(ImageAnalysisAbstractAnalyzer.java:286)
    at androidx.camera.core.ImageAnalysisAbstractAnalyzer$$ExternalSyntheticLambda0.run(D8$$SyntheticClass:0)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1137)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:637)
    at java.lang.Thread.run(Thread.java:1012)

请问为何使用Mutex仍会触发ConcurrentModificationException，该如何解决这个问题？

问题根源 #

你遇到的ConcurrentModificationException并非多协程并发读写导致，而是在遍历ArrayList的同时直接修改其结构触发的。ArrayList的迭代器采用"快速失败"机制：当迭代器创建后，如果集合的结构被修改（比如调用remove、add），迭代器的next()方法就会抛出该异常。你的Mutex已经保证了同一时间只有一个协程执行removeUnusedTrains方法，但遍历过程中直接调用trains.remove(train)的操作破坏了迭代器的一致性，所以依然会抛出异常。

修复方案 #

以下是几种可行的修复方式：

方案1：使用迭代器的remove方法 #

通过迭代器遍历并删除元素，迭代器的remove()方法会同步更新迭代器状态，避免触发异常：

suspend fun removeUnusedTrains() = mutex.withLock {
    val iterator = trains.iterator()
    while (iterator.hasNext()) {
        val train = iterator.next()
        if (!train.changed) {
            iterator.remove()
        } else {
            train.changed = false
        }
    }
    Log.d("TrainService", "Trains now: ${trains.joinToString()}")
}

方案2：先收集待删除元素，再批量删除 #

先遍历收集所有需要删除的Train实例，再一次性从列表中移除，避免遍历过程中修改集合结构：

suspend fun removeUnusedTrains() = mutex.withLock {
    val trainsToRemove = trains.filter { !it.changed }
    trains.removeAll(trainsToRemove)
    trains.forEach { it.changed = false }
    Log.d("TrainService", "Trains now: ${trains.joinToString()}")
}

方案3：使用removeIf方法（最简洁） #

利用Kotlin集合的removeIf方法批量删除符合条件的元素，再统一重置剩余元素的状态：

suspend fun removeUnusedTrains() = mutex.withLock {
    trains.removeIf { !it.changed }
    trains.forEach { it.changed = false }
    Log.d("TrainService", "Trains now: ${trains.joinToString()}")
}

内容的提问来源于stack exchange，提问作者Filip Černák