RxJava ConcurrentModificationException

Question

我是RxJava的新手但总有一天..。；；)

我所拥有的：

• 用事件计划器模型接收json
• 在此模型中，are任务
• 我希望将此任务保存到本地db中，但也希望在发生关系时向它们添加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)

这就是如何保存到db中的方式：

    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。