使用 iter.hasNext() 时出现 ConcurrentModificationException [关闭]

Question

我有一个producer consumer 设置，使用Wildfly AS for JMS，生产者每1分钟使用一次newFixedThreadPool(126)，每个线程从REST服务中提取数据并推送它到HornetQWildfly AS 上。

然后在消费者方面，我有一个消费者类，它使用 HornetQ 中的消息和一个简单的 Parser 类，对于数据库插入，我正在尝试缓冲数据库插入并且我得到一个异常在线程“main”中 java.util.ConcurrentModificationException 我怀疑这与我的代码不是线程安全有关，但我无法缩小范围。

消费者：

    public void Consume(Consumer asyncReceiver) throws Throwable {

        try (javax.jms.Connection connection = connFactory.createConnection(props.getProperty("DEFAULT_USERNAME"), props.getProperty("
             Session queueSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

             MessageConsumer msgConsumer = queueSession.createConsumer(queue)) {

            msgConsumer.setMessageListener(asyncReceiver);

            connection.setExceptionListener(asyncReceiver);

            connection.start();

            /** I think this is causing the problem */
            System.out.println("waiting for messages");
            int bufferCount = 0;
            for (int i = 0; i < 47483647; i++) {
                Thread.sleep(1000);
                System.out.print(".");
                if (bufferCount == 5) {
                    if(responseList.size() > 50){
                        this.buffer();
                    }
                    bufferCount = 0;
                }
                bufferCount++;
            }
            System.out.println();
        }

    } catch (Exception e) {
        log.severe(e.getMessage());
        throw e;
    } finally {
        if (context != null) {
            context.close();
        }
    }
    if (connection != null) {
        connection.close();
    }
}

public void buffer() throws Exception {
        System.out.println("Parsing: " + responseList.size() + " messages");
        Parser parser = new Parser();
        parser.addList(responseList);
        parser.parseApplication();
        responseList.clear();
}

@Override
public void onMessage(Message message) {
    TextMessage msg = (TextMessage) message;
    try {
        responseList.add(msg.getText());
    } catch (Exception e) {
        e.printStackTrace();
    }

解析器：

public class Parser {

private ArrayList<String> responseList;


public void addList(ArrayList<String> list) {
    this.responseList = list;
}

public void parseApplication() throws Exception {

    DBConnection db = DBConnection.createApplication();

    try (Connection connection = db.getConnection()) {

//Code removed for stack question

        Iterator<String> iter = responseList.iterator();
        while(iter.hasNext()) {

// This is where the error is thrown 
            while (fieldsIterator.hasNext()) {

            //Cut code from here, basic JSON parsing done here
               timeslices = parse(iter.Next())

                for (int i = 0; i < timeslices.length(); i++) {

                        ThroughputEntry TP = new ThroughputEntry();
                        TP.setThroughput(values.getDouble(name));
                        TP.setEnvironment(envName);
                        TP.setName(appName);
                        TP.setRetrieved(from);
                        TP.setPeriodEnd(to);
                        db.addHistory(TP);
                    }
                }
            }
        }
        iter.remove();
    }
}
}

也许我应该把我的缓冲方法变成threadpool？

Answer 1

responseList.add(msg.getText());

在您的解析器迭代列表时修改列表。 在这种情况下，迭代器会抛出 ConcurrentModificationException。 要修复它，您应该在解析之前复制列表。