java - 抛出 ConcurrentModificationException(内部有多个 for 循环)

Question

很抱歉这不是 SSCCE，但我已尝试充分描述此问题中的对象。

数据结构

嵌套在里面的都是成员变量

PerformanceData (Object)
|- process (String): The name of the process this PerformanceData has data for (ie: firefox.exe)
|- metrics (Map<String, List<DataValue>>): The keys are metrics whose usage we are monitoring ("cpu", "disk", "memory")

DataValue (Object)
|- value (double): The observed value (ie: 0.43 which means 43% cpu usage)
|- time (Date) : The time the value was observed

ModelEnsemble (Thread)
|- data (List<DataValue>): The DataValues available to this ModelEnsemble
|- models (Map<String, IEnsembleModel>): maps a name to each IEnsembleModel.
||-This is used because the user can choose which IEnsembleModels to run via a myApp.properties file.
||-In the constructor, we parse this properties file for enabled IEnsembleModels

IEnsembleModel (Object)
|- window (int): The max size input should be. When input.size() > window, we remove (older elements) from the front of the queue
|- input (ArrayDeque<DataValue>): The queue of DataValues we are modelling
|- addInput (DataValue): adds a DataValue to the end of the list
|- getLastInput (DataValue): The last inserted DataValue (from addInput)
|- getLastPrediction (double): The second latest predicted outcome from the current input data (minus the last input)
|- getError (double): Percent error, how much the last prediction (getLastPrediction) differed from the last input (getLastInput)
|- getNextPrediction (double): The latest predicted outcome from the current input data (including the last input)
|- model (double): Computes the next prediction, returns getNextPrediction().

问题

以下代码抛出 ConcurrentModificationException。我相信这是由匿名 PropertChangeListener 以及 addInput() 引起的。一些慷慨的 SO 用户可以查看我的代码并指出我允许多次编辑或竞争条件的地方吗？

外(主)循环:

for (final PerformanceData perfData : myPerformanceData) {
    for (Map.Entry<String, List<DataValue>> entry : perfData.getMetrics().entrySet()) {

        final String metric             = entry.getKey();
        final List<DataValue> values    = entry.getValue();

        ensemble = new ModelEnsemble(values);

        // Pretty sure this causes the Concurrent Modification Exception
        ensemble.addPropertyChangeListener(new PropertyChangeListener() {

            @Override
            public void propertyChange(PropertyChangeEvent evt) {

                IEnsembleModel model = (IEnsembleModel) evt.getNewValue();
                ModelPanel.this.firePropertyChange("result", null, new ModelResult(
                        perfData.getProcess(),
                        metric,
                        model.getLastInput().getValue(),
                        model.getLastPrediction(),
                        model.getError()
                ));
            }
        });

        ensemble.start();
    }
}

ModelEnsemble 的 run() 循环

int offset = 0; // array index we are currently modelling
while (!this.isInterrupted()) {


    if (offset > data.size() - 1) {

        // we've exhausted the input input, close the model solver
        System.out.println("Input input exhausted, ending ensemble");
        this.interrupt();

    } else {

        // get the latest input value
        DataValue value = data.get(offset);

        for (Map.Entry<String, IEnsembleModel> entry : models.entrySet()) {
            String name         = entry.getKey();
            IEnsembleModel model= entry.getValue();

            model.addInput(value);
            model.model();
            this.notifyListeners(name, model); // triggers that anonymous PropertyChangeListener in the above piece of code
        }
    }

    offset++; // so we can model the next element of the DataValue List
}

IEnsembleModel 示例方法

double model() {
    double sum = 0;

    for (DataValue value : input) {
        sum += value.getValue();
    }

    setNextPrediction(sum / ((double) input.size()));
    return getNextPrediction();
}

void addInput(DataValue input) {
    this.input.addLast(input); // add to back of queue

    this.maintainWindow();
}

private void maintainWindow() {
    int size = this.input.size(); 
    while (size > window) { // by default window = 10
        this.input.pop();   // error here
    }
}

堆栈跟踪

这发生在每个线程中:

Exception in thread "Thread-13" java.util.ConcurrentModificationException
    at java.util.ArrayDeque$DeqIterator.next(ArrayDeque.java:632)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.maintainWindow(MovingAverageModel.java:93)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.addInput(MovingAverageModel.java:53)
    at ca.yorku.cirillom.ensemble.models.ModelEnsemble.run(ModelEnsemble.java:123)

这也发生在每个线程中:

Exception in thread "Thread-7" java.util.NoSuchElementException
    at java.util.ArrayDeque.removeFirst(ArrayDeque.java:278)
    at java.util.ArrayDeque.pop(ArrayDeque.java:507)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.maintainWindow(MovingAverageModel.java:85)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.addInput(MovingAverageModel.java:49)
    at ca.yorku.cirillom.ensemble.models.ModelEnsemble.run(ModelEnsemble.java:123)

我觉得某处一定存在竞争条件，因为 maintainWindow() 检查 input.size() 是否大于 window，默认情况下为 10。如果 size() 大于 10

，则不会发生 NoSuchElementException

Answer 1

只有在以下情况下才能抛出并发修改异常:

1. 您正在使用迭代器或 foreach 迭代循环
2. 您在迭代器仍在运行时从列表中添加或删除某些内容。

这两行正在进行迭代，因此在这些 for 循环退出之前修改 myPerformanceData 或 perfData.getMetrics。

for (final PerformanceData perfData : myPerformanceData) {
     for (Map.Entry<String, List<DataValue>> entry : perfData.getMetrics().entrySet()) {

或者这里的这一行:

    for (Map.Entry<String, IEnsembleModel> entry : models.entrySet()) {

错误的完整堆栈跟踪应该为您提供并发修改发生的位置，并让您识别正在修改的集合。然后，您只需要修改您的逻辑，以便在迭代它们之外发生对集合的更改。