java - RxJava 并发修改异常

Question

我是 RxJava 的菜鸟，但总有一天……； )

我有什么:

• 使用 Event Planner 模型接收 json
• 在此模型中有任务
• 我想将此任务保存到本地数据库中，但也想在关系的情况下添加到它们的 parent_id

有经验的人可以看看这段代码吗:

    private fun saveEventPlanners(eventPlanners: ArrayList<EventPlanner>) {
    LogMgr.d(TAG, "saveEventPlanners() events:$eventPlanners")

    compositeDisposable.add(wfmStorageDomain.saveEventPlanners(eventPlanners)
            .subscribeOn(rxSchedulers.computation())
            .observeOn(rxSchedulers.computation())
            .subscribe({ saved ->
                LogMgr.v(TAG, "event planners saved successfully, value: $saved")

                val taskList = ArrayList<Task>()

                eventPlanners.forEach { eventPlanner ->

                    if (eventPlanner.status == EventPlanner.EventPlannerStatus.NEW) {
                        FS.get().wfmComponent.getEventPlannerStatusChanger().updateEventPlannersStatus(eventPlanner.event_id!!, EventPlanner.EventPlannerStatus.RECEIVED)
                    }

                    (eventPlanner.tasks as ArrayList<Task>).forEach {
                        val task = Task()
                        task.id = it.id
                        task.status_id = Task.STATUS.NEW
                        task.name = it.name
                        task.end_scenario_id = it.end_scenario_id
                        task.start_scenario_id = it.start_scenario_id
                        task.isBind = it.isBind
                        task.parent_event_planner_id = eventPlanner.event_planner_id

                        taskList.add(task)
                    }
                    compositeDisposable.add(wfmStorageDomain.saveTasks(taskList)
                            .subscribeOn(rxSchedulers.computation())
                            .observeOn(rxSchedulers.computation())
                            .subscribe({
                                LogMgr.d(TAG, "tasks saved successfully = $it")
                            }, {
                                LogMgr.e(TAG, "Error while saving tasks", it)
                            }))

                }
            }, {
                LogMgr.e(TAG, "Error while saving event planners", it)
            }))

    notifyObservers(eventPlanners)
}

我在这个函数中遇到错误:

java.util.ConcurrentModificationException
    at java.util.ArrayList$ArrayListIterator.next(ArrayList.java:573)
    at com.raizlabs.android.dbflow.sql.saveable.ListModelSaver.saveAll(ListModelSaver.java:32)
    at com.raizlabs.android.dbflow.sql.saveable.ListModelSaver.saveAll(ListModelSaver.java:19)
    at com.raizlabs.android.dbflow.structure.ModelAdapter.saveAll(ModelAdapter.java:196)
    at com.raizlabs.android.dbflow.rx2.structure.RXModelAdapter$3.call(RXModelAdapter.java:61)
    at com.raizlabs.android.dbflow.rx2.structure.RXModelAdapter$3.call(RXModelAdapter.java:58)
    at io.reactivex.internal.operators.completable.CompletableFromCallable.subscribeActual(CompletableFromCallable.java:35)
    at io.reactivex.Completable.subscribe(Completable.java:1919)
    at io.reactivex.internal.operators.completable.CompletableSubscribeOn$SubscribeOnObserver.run(CompletableSubscribeOn.java:64)
    at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:579)
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
    at java.util.concurrent.FutureTask.run(FutureTask.java:234)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:153)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:267)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1080)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:573)
    at java.lang.Thread.run(Thread.java:841)

这是保存到数据库中的样子:

    override fun saveTasks(tasks: ArrayList<Task>): Single<Boolean> {
    LogMgr.d(TAG, "saveTasks() : $tasks")
    return Single.create({ emitter ->
        RXModelAdapter.from(Task::class.java)
                .saveAll(tasks)
                .subscribeOn(getSubscriptionScheduler())
                .subscribe({
                    LogMgr.d(TAG, "saveTasks() onComplete")
                    emitter.onSuccess(true)

                }, {
                    LogMgr.e(TAG, "saveTasks() onError ", it)
                    emitter.onError(it)
                })
    })
}

  private fun getSubscriptionScheduler(): Scheduler {
    return FS.get().commonComponent.rxSchedulers.get().sqlite()
}

Answer 1

问题是这样的:

val taskList = ArrayList<Task>()

eventPlanners.forEach { eventPlanner ->

    // ...

    compositeDisposable.add(wfmStorageDomain.saveTasks(taskList)

    // ...
}

您有一个列表，您不断从不同的 eventPlanner 项目添加该列表并将该部分列表提交给保存任务。如果 eventPlanners 中有多个项目，这将导致 ConcurrentModificationException 错误。将 taskList 设为 forEach 循环的本地，或者将 saveTasks 移出 forEach。