Mutex 不会停止多个协程来修改或读取数据

Question

我有一个包含列车列表的服务（我想在协程之间共享的重要数据）。它还具有修改列表的方法。因此，我在这些方法中使用互斥体。 这是调用方法的代码（位于协程内部）：

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)

这是火车服务

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)

Answer 1

问题不是来自互斥锁，而是来自于您在 foreach 循环中调用删除这一事实。列表在浏览时修改，不支持。

为了使您的删除功能正常工作，您应该使用listIterator，它支持在循环的同时进行修改：

    suspend fun removeUnusedTrains() = mutex.withLock {
        val iter = traines.listIterator()
        while (iter.hasNext()) {
            val train = iter.next()
            if (!train.changed) {
                // Ask iterator to remove current element from the list
                iter.remove()
            } else {
                train.changed = false
            }
        }
        Log.d("TrainService", "Trains now: ${trains.joinToString()}")
    }