7

我正在构建一个使用 Spring @EnableAsync 和 @Async 注释的多线程 Spring Boot 应用程序。当我使用单线程（CorePoolSize 1，MaxPoolSize 1）运行应用程序时，一切都按预期工作。当我在似乎随机出现的情况下将池大小增加到 1 以上时，我得到java.sql.SQLNonTransientException: [Amazon]JDBC Not all parameters have been packed。调用 Amazon AWS Redshift 数据库时出错。

在 ServiceProcessBean.java 中，我自动连接了我的 ProcessService 类（要完成的线程工作）和 ShipmentDAO，它加载了要由 ProcessService.process() 方法处理的装运 ID 列表，代码如下。

@Component
public class ShipmentBatchBean {

  private Logger logger = LoggerFactory.getLogger(this.getClass());

  @Autowired
  private ShipmentDAO shipmentDAO;

  @Autowired
  private ProcessService processService;

  @Scheduled(
    initialDelayString = "${executor.delay.initial}",
    fixedDelayString = "${executor.delay.fixed}"
  )
  public void cronJob() throws InterruptedException, ExecutionException {

    List<CompletableFuture<Boolean>> asyncResponse = new ArrayList<>();

    logger.info("Starting cronJob() method");

    try {
      List<String> shipments = shipmentDAO.getAllShipmentsReadyForIeta();
      logger.info("There are {} shipments to be processed in the data store", 
          shipments.size());

      for(String shipment : shipments) {
        asyncResponse.add(processService.process(shipment));
      }
    } catch (Exception ex) {
      logger.error(ex.getMessage());
      ex.printStackTrace();
    }

    CompletableFuture.allOf(asyncResponse.toArray(
      new CompletableFuture[asyncResponse.size()]
    )).join();

    logger.info("Ending cronJob() method");
  }

}

最后，在 ProcessService 中，我们自动连接多个存储库和一个 JSON 服务，并使用 @Async 注释开始 process() 方法。请参阅下面的代码片段。

@Service
public class ProcessServiceBean implements ProcessService {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private ShipmentDAO shipmentDAO;

    @Autowired
    private OssItemDAO ossItemDAO;

    @Autowired
    private OssHeaderDAO ossHeaderDAO;

    @Autowired
    private OssDataJsonServiceBean ossDataJsonServiceBean;

    @Override
    @Async
    public CompletableFuture<Boolean> process(String shipmentId) {

      Shipment shipment = null;

      logger.debug("Retrieving from ieta_input (shipment id {})",
        shipmentId);
      try {
        shipment = shipmentDAO.getOneBy(shipmentId);
      } catch (SQLException e) {
        logger.error("process of shipment {}) ended in error",
          shipmentId,
      };
      return CompletableFuture.completedFuture(false);
    }

 code snipped for brevity.

最后，在 ShipmentDAO 中，我们有 getOneBy() 方法，它返回请求的货运记录。

@Override
public Shipment getOneBy(String shipmentId) throws SQLException {
    Shipment shipment = null;
    Connection conn = null;

    String sql = "SELECT * FROM myschema.tablename WHERE shipmentid = ? LIMIT 1";

    try {
      conn = dataSource.getConnection();
      PreparedStatement ps = conn.prepareStatement(sql);
      ps.setString(1, shipmentId);
      ResultSet rs = ps.executeQuery();
      if(rs.next()) {
        shipment = new Shipment(
            rs.getLong("rowid"),
            rs.getString("shipmentid"),
            ...
            rs.getString("more_info_here")
        );
      }
      ps.close();
    } catch(SQLException e) {
      logger.error("ShipmentDAO.getOneBy() failed: {}",
        e.getMessage());
      e.printStackTrace();
      throw e;
    } finally {
      if(conn != null) {
      try {
        conn.close();
      } catch (SQLException se) { }
    }
  }
  return shipment;
}

我已经联系了 AWS，他们的知识库中只有两个实例，其中报告了这个错误，只有“用户代码问题”是解决方案。

我没有看到任何明显的线程安全问题，但它的行为就像这样，因为当只触发一个线程并且每批货物都按顺序处理时，代码可以完美地工作。

有没有人看到明显的问题或需要更多信息？任何帮助表示赞赏。

堆栈跟踪如下：

2018-Oct-22 15:57:30.960 ERROR- [shipment-executor-8] [Amazon][JDBC](10900) Not all parameters have been populated.
org.springframework.dao.InvalidDataAccessResourceUsageException: could not extract ResultSet; SQL [n/a]; nested exception is org.hibernate.exception.SQLGrammarException: could not extract ResultSet
at org.springframework.orm.jpa.vendor.HibernateJpaDialect.convertHibernateAccessException(HibernateJpaDialect.java:261)
at org.springframework.orm.jpa.vendor.HibernateJpaDialect.translateExceptionIfPossible(HibernateJpaDialect.java:244)
at org.springframework.orm.jpa.AbstractEntityManagerFactoryBean.translateExceptionIfPossible(AbstractEntityManagerFactoryBean.java:503)
at org.springframework.dao.support.ChainedPersistenceExceptionTranslator.translateExceptionIfPossible(ChainedPersistenceExceptionTranslator.java:59)
at org.springframework.dao.support.DataAccessUtils.translateIfNecessary(DataAccessUtils.java:209)
at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:147)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.data.jpa.repository.support.CrudMethodMetadataPostProcessor$CrudMethodMetadataPopulatingMethodInterceptor.invoke(CrudMethodMetadataPostProcessor.java:133)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:92)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.data.repository.core.support.SurroundingTransactionDetectorMethodInterceptor.invoke(SurroundingTransactionDetectorMethodInterceptor.java:57)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
at com.sun.proxy.$Proxy91.getFirstByFactDeliveryNumberIsOrSapDeliveryNumberIs(Unknown Source)
at com.accenture.service.ProcessServiceBean.process(ProcessServiceBean.java:91)
at com.accenture.service.ProcessServiceBean$$FastClassBySpringCGLIB$$45e1a1ec.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:738)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:99)
at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:282)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.aop.interceptor.AsyncExecutionInterceptor$1.call(AsyncExecutionInterceptor.java:115)
at org.springframework.aop.interceptor.AsyncExecutionAspectSupport$CompletableFutureDelegate$1.get(AsyncExecutionAspectSupport.java:328)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.hibernate.exception.SQLGrammarException: could not extract ResultSet
at org.hibernate.exception.internal.SQLStateConversionDelegate.convert(SQLStateConversionDelegate.java:106)
at org.hibernate.exception.internal.StandardSQLExceptionConverter.convert(StandardSQLExceptionConverter.java:42)
at org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:109)
at org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:95)
at org.hibernate.engine.jdbc.internal.ResultSetReturnImpl.extract(ResultSetReturnImpl.java:79)
at org.hibernate.loader.Loader.getResultSet(Loader.java:2117)
at org.hibernate.loader.Loader.executeQueryStatement(Loader.java:1900)
at org.hibernate.loader.Loader.executeQueryStatement(Loader.java:1876)
at org.hibernate.loader.Loader.doQuery(Loader.java:919)
at org.hibernate.loader.Loader.doQueryAndInitializeNonLazyCollections(Loader.java:336)
at org.hibernate.loader.Loader.doList(Loader.java:2617)
at org.hibernate.loader.Loader.doList(Loader.java:2600)
at org.hibernate.loader.Loader.listIgnoreQueryCache(Loader.java:2429)
at org.hibernate.loader.Loader.list(Loader.java:2424)
at org.hibernate.loader.hql.QueryLoader.list(QueryLoader.java:501)
at org.hibernate.hql.internal.ast.QueryTranslatorImpl.list(QueryTranslatorImpl.java:371)
at org.hibernate.engine.query.spi.HQLQueryPlan.performList(HQLQueryPlan.java:216)
at org.hibernate.internal.SessionImpl.list(SessionImpl.java:1326)
at org.hibernate.internal.QueryImpl.list(QueryImpl.java:87)
at org.hibernate.jpa.internal.QueryImpl.list(QueryImpl.java:606)
at org.hibernate.jpa.internal.QueryImpl.getSingleResult(QueryImpl.java:529)
at org.hibernate.jpa.criteria.compile.CriteriaQueryTypeQueryAdapter.getSingleResult(CriteriaQueryTypeQueryAdapter.java:54)
at org.springframework.data.jpa.repository.query.JpaQueryExecution$SingleEntityExecution.doExecute(JpaQueryExecution.java:208)
at org.springframework.data.jpa.repository.query.JpaQueryExecution.execute(JpaQueryExecution.java:87)
at org.springframework.data.jpa.repository.query.AbstractJpaQuery.doExecute(AbstractJpaQuery.java:116)
at org.springframework.data.jpa.repository.query.AbstractJpaQuery.execute(AbstractJpaQuery.java:106)
at org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.doInvoke(RepositoryFactorySupport.java:499)
at org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.invoke(RepositoryFactorySupport.java:477)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.data.projection.DefaultMethodInvokingMethodInterceptor.invoke(DefaultMethodInvokingMethodInterceptor.java:56)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:99)
at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:282)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:136)
... 24 more
Caused by: java.sql.SQLNonTransientException: [Amazon][JDBC](10900) Not all parameters have been populated.
at com.amazon.exceptions.ExceptionConverter.toSQLException(Unknown Source)
at com.amazon.jdbc.common.SPreparedStatement.executeQuery(Unknown Source)
at sun.reflect.GeneratedMethodAccessor79.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.tomcat.jdbc.pool.StatementFacade$StatementProxy.invoke(StatementFacade.java:114)
at com.sun.proxy.$Proxy101.executeQuery(Unknown Source)
at org.hibernate.engine.jdbc.internal.ResultSetReturnImpl.extract(ResultSetReturnImpl.java:70)
... 55 more
4

2 回答 2

14

我们终于有决心了！经过一组艰苦的测试后，它归结为 JDBC 驱动程序不是线程安全的。我们在使用多个版本的 redshift-jdbc41 和 redshift-jdbc42 驱动程序时都遇到了这个问题，最终确定了一个<version>1.2.10.1009</version>不再抛出错误并同时处理多个线程的版本。如果您在 Amazon Support 处有一张未处理的票，您可以将他们转至案例 ID 5466870321，我们在其中提供了所有必要的详细信息，以便他们确认问题出在驱动程序本身。

正如代码 tachyon所指出的（感谢您的输入），问题似乎出在 JDBC 驱动程序的准备好的语句功能中，因此我们没有看到数据库本身因这些问题而导致任何数据损坏，因为它们在请求之前持续存在被转发到 Redshift。

最后一点，我非常惊讶亚马逊发布了相当多的 JDBC 驱动程序版本，我们在 1.2.10.1009 之后测试的所有内容都包含这个线程安全问题，但亚马逊并不知道，也没有单个堆栈溢出或其他谷歌搜索表明这是一个问题，但这篇文章除外。

我希望这可以帮助那些发现他们的多线程应用程序无法正常工作的人。

于 2018-11-03T13:13:39.890 回答
2

遇到了一个类似的问题，cuplrit 是在使用命名参数时在服务器端准备语句缓存。准备好的语句和命名参数是单独发送的，redshift jdbc 驱动程序无法解析跨多个并发连接发送的同一查询的命名参数。临时解决方法是替换“？” 在 "String sql = "SELECT * FROM myschema.tablename WHERE shippingid = ? LIMIT 1"; 使用实际参数。这当然会使您面临 sql 注入攻击，并且您将失去具有执行计划的缓存预准备语句的好处。但是，我们决定继续使用该选项，直到永久修复可用，YMMV。

于 2018-10-31T22:53:48.130 回答