ThreadPoolTask​​Executor 中的 ConcurrentModificationException 错误

Question

我使用这样的配置注释创建 ThreadPoolTask​​Executor

public class AsyncConfiguration implements AsyncConfigurer, EnvironmentAware {

    private final Logger log = LoggerFactory.getLogger(AsyncConfiguration.class);

    private RelaxedPropertyResolver propertyResolver;

    @Override
    public void setEnvironment(Environment environment) {
        this.propertyResolver = new RelaxedPropertyResolver(environment, "async.");
    }

    @Override
    @Bean
    public Executor getAsyncExecutor() {
        log.debug("Creating Async Task Executor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(propertyResolver.getProperty("corePoolSize", Integer.class, 30));
        executor.setMaxPoolSize(propertyResolver.getProperty("maxPoolSize", Integer.class, 150));
        executor.setQueueCapacity(propertyResolver.getProperty("queueCapacity", Integer.class, 10000));
        executor.setThreadNamePrefix("app-Executor-");
        return new ExceptionHandlingAsyncTaskExecutor(executor);
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
}

我在这个类中使用 ThreadPoolExecutor

@Component
public class TrapReceiver extends Thread implements CommandResponder {

    @Inject
    private ApplicationContext applicationContext;

    @Inject
    private Executor executor;

    public TrapReceiver(){
    }

    List<PDUv1> listPdu = new ArrayList<PDUv1>();

    @PostConstruct
    public void init() {
        this.start();
    }

    public synchronized void processPdu(CommandResponderEvent cmdRespEvent) {
        PDUv1 pdu = (PDUv1) cmdRespEvent.getPDU();
        listPdu.add(pdu);
        if (pdu != null) {
            if(listPdu.size() == 3){ //3 pdu per thread
                List<PDUv1> temp = new ArrayList<PDUv1>();
                temp.addAll(listPdu);
                TrapInsertor trapInsertor = (TrapInsertor) applicationContext.getBean("trapInsertor");
                trapInsertor.setProperty(temp);
                executor.execute(trapInsertor);
                listPdu.clear();
            }
        }
    }

这是我的线程类

public class TrapInsertor implements Runnable {

    @Inject
    private TrapProcessorService trapProcessorService;

    private List<PDUv1> listPdu;

    public void setProperty(List<PDUv1> listPdu){
        this.listPdu = listPdu;
    }

    @Override
    public void run() {
        try{
            System.out.println(Thread.currentThread().getName()+" Start process "+listPdu.size()+" PDU");
            for(PDUv1 pdu : listPdu){
                String[] varBinding = pdu.getVariableBindings().toString().replace("[", "").replace("]", "").split(", ");
                trapProcessorService.processTrap(varBinding);
            }
            listPdu.clear();
        }catch (Exception e) {
            e.printStackTrace();
        }finally{
        }
    }
}

但有时我会遇到这样的错误

java.util.ConcurrentModificationException
    at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:859)
    at java.util.ArrayList$Itr.next(ArrayList.java:831)
    at app.snmp.test.TrapInsertor.run(TrapInsertor.java:37)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

也喜欢这个

org.springframework.dao.CannotAcquireLockException: could not execute statement; SQL [n/a]; nested exception is org.hibernate.exception.LockAcquisitionException: could not execute statement
    at org.springframework.orm.jpa.vendor.HibernateJpaDialect.convertHibernateAccessException(HibernateJpaDialect.java:239)
    at org.springframework.orm.jpa.vendor.HibernateJpaDialect.translateExceptionIfPossible(HibernateJpaDialect.java:214)
    at org.springframework.orm.jpa.JpaTransactionManager.doCommit(JpaTransactionManager.java:521)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:757)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:726)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:497)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:277)
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:653)
    at app.snmp.test.service.TrapProcessorService$$EnhancerBySpringCGLIB$$3d040253.processTrap(<generated>)
    at app.snmp.test.TrapInsertor.run(TrapInsertor.java:39)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.hibernate.exception.LockAcquisitionException: could not execute statement
    at org.hibernate.dialect.MySQLDialect$1.convert(MySQLDialect.java:451)
    at org.hibernate.exception.internal.StandardSQLExceptionConverter.convert(StandardSQLExceptionConverter.java:49)
    at org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:126)
    at org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:112)
    at org.hibernate.engine.jdbc.internal.ResultSetReturnImpl.executeUpdate(ResultSetReturnImpl.java:211)
    at org.hibernate.engine.jdbc.batch.internal.NonBatchingBatch.addToBatch(NonBatchingBatch.java:62)
    at org.hibernate.persister.entity.AbstractEntityPersister.update(AbstractEntityPersister.java:3281)
    at org.hibernate.persister.entity.AbstractEntityPersister.updateOrInsert(AbstractEntityPersister.java:3183)
    at org.hibernate.persister.entity.AbstractEntityPersister.update(AbstractEntityPersister.java:3525)
    at org.hibernate.action.internal.EntityUpdateAction.execute(EntityUpdateAction.java:159)
    at org.hibernate.engine.spi.ActionQueue.executeActions(ActionQueue.java:463)
    at org.hibernate.engine.spi.ActionQueue.executeActions(ActionQueue.java:349)
    at org.hibernate.event.internal.AbstractFlushingEventListener.performExecutions(AbstractFlushingEventListener.java:350)
    at org.hibernate.event.internal.DefaultFlushEventListener.onFlush(DefaultFlushEventListener.java:56)
    at org.hibernate.internal.SessionImpl.flush(SessionImpl.java:1222)
    at org.hibernate.internal.SessionImpl.managedFlush(SessionImpl.java:425)
    at org.hibernate.engine.transaction.internal.jdbc.JdbcTransaction.beforeTransactionCommit(JdbcTransaction.java:101)
    at org.hibernate.engine.transaction.spi.AbstractTransactionImpl.commit(AbstractTransactionImpl.java:177)
    at org.hibernate.jpa.internal.TransactionImpl.commit(TransactionImpl.java:77)
    at org.springframework.orm.jpa.JpaTransactionManager.doCommit(JpaTransactionManager.java:517)
    ... 12 more
Caused by: com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
    at com.mysql.jdbc.Util.handleNewInstance(Util.java:377)
    at com.mysql.jdbc.Util.getInstance(Util.java:360)
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:985)
    at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3887)
    at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3823)
    at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2435)
    at com.mysql.jdbc.ServerPreparedStatement.serverExecute(ServerPreparedStatement.java:1288)
    at com.mysql.jdbc.ServerPreparedStatement.executeInternal(ServerPreparedStatement.java:794)
    at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:2141)
    at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:2077)
    at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:2062)
    at com.zaxxer.hikari.proxy.PreparedStatementProxy.executeUpdate(PreparedStatementProxy.java:61)
    at com.zaxxer.hikari.proxy.PreparedStatementJavassistProxy.executeUpdate(PreparedStatementJavassistProxy.java)
    at org.hibernate.engine.jdbc.internal.ResultSetReturnImpl.executeUpdate(ResultSetReturnImpl.java:208)
    ... 27 more

我的代码有什么问题？如果 listPdu 集合引起的错误在线程之间共享？ ........

ConcurrentModificationException 错误通过在这个配置类中添加作用域原型来解决。原型意味着每次请求时都会创建新的 bean 实例。谢谢@Vladimir Sitnikov

@Configuration
public class TrapProcessorComponentConfiguration {

    @Bean
    @Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public TrapInsertor trapInsertor(){
        return new TrapInsertor();
    }
}

Answer 1

问题出在您的弹簧配置中。您可能会重复使用相同的 trapInsertor 实例，因此会出现所有异常。

您的“trapInsertor”bean 是单例 bean 吗？我认为 bean 在 spring 中默认是单例的。

看看会发生什么：

1. 你收集了 3 件物品
2. 调用trapInsertor.setProperty(temp); // 说它是arrayList1
3. 调用executor.execute(trapInsertor);，但trapInsertor 尚未启动（线程池可能需要一段时间才能开始工作）
4. 你又收集了 3 件物品
5. 调用trapInsertor.setProperty(temp); // 说它是arrayList2
6. 致电executor.execute(trapInsertor);
7. 现在#3 和#6 的动作开始起作用了。他们可能会看到相同的arrayList2。

所以你可以拥有：

1. 数据丢失。基本上，arrayList1 根本不被处理！
2. 一个线程遍历列表for(PDUv1 pdu : listPdu){（在TrapInsertor）和另一个执行listPdu.clear()（在TrapInsertor）。这导致ConcurrentModificationException。
3. 两个线程都可能将数据写入数据库，从而相互锁定，因此MySQLTransactionRollbackException: Deadlock found when trying to get lock;

我建议你应该重新创建你提交给执行者的任务。 换句话说，您需要类似scope="prototype" 的东西（参见How do I force a spring container not to return a singleton instance of a bean?）

Answer 2

您可以制作正在处理的列表的防御性副本，只需将TrapInsertor.setProperty() 修改为：

public void setProperty(List<PDUv1> listPdu){ this.listPdu = new ArrayList<PDUv1>(listPdu); }

这样，每个线程将使用自己的列表实例，从而防止并发修改。