Spring Batch Reader 的游标在 JTA 事务管理步骤中提前关闭

Question

相关步骤的工作配置如下：

• Step、Spring Batch Job Repository 和业务存储库（使用各种数据源）都使用 JTA 事务管理器。
• 步骤“myStep”使用 Jdbc Paging Item Reader。
• WebLogic、Oracle XE 和/或 EE

我想在“myStep”中分析 Jdbc Cursor Item Reader 的性能，但是在第一次提交后，第二个块的第一次读取会失败并出现 java.sql.SQLException：结果集已经关闭。

我怀疑可能是 JTA / XA 驱动程序出于某种原因关闭了游标，所以我给“myStep”一个简单的数据源事务管理器（在读者正在使用的数据源上），并且该步骤能够成功完成。这不是解决方案，因为这会破坏步骤的事务完整性。

我应该能够在 JTA 托管步骤中使用光标阅读器（使用下面描述的环境）吗？如果是这样，我这边可能配置不正确？

环境

• 事务经理：<bean id="myTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>
• 数据源驱动程序：OracleXADataSource JDBC 6 11.1.0.7.0
• WebLogic：12.1.3.0.0
• Oracle DB 11g：企业版 11.2.0.4.0
• 操作系统：OSX 或 Linux

配置

<bean id="myTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>

<bean id="myDataSource" class="org.springframework.jndi.JndiObjectFactoryBean">
    <property name="jndiName" value="jdbc/myDataSource"/>
    <property name="proxyInterface" value="javax.sql.DataSource"/>
</bean>

<batch:step id="myStep" job-repository="myJobRepositoryFactory">
    <batch:tasklet transaction-manager="myTransactionManager">
        <batch:chunk
                reader="myReader"
                processor="myProcessor"
                writer="myWriter"
                commit-interval="100"
                processor-transactional="false"/>
        <batch:listeners>
            <batch:listener ref="myListener"/>
        </batch:listeners>
    </batch:tasklet>
</batch:step>

<bean id="myReader" class="org.springframework.batch.item.database.JdbcCursorItemReader" scope="step">
    <property name="dataSource" ref="myDataSource"/>
    <property name="sql" value="SELECT * FROM myHugeTable ORDER BY myColumn DESC"/>
    <property name="rowMapper">
        <bean class="myRowMapper"/>
    </property>
</bean>

陷入困境

下面是在读取下一个块之前关闭的结果集的调用堆栈。注意 XA 连接关闭所有语句，这会导致 JDBC 关闭所有结果集。

java.lang.Thread.State: RUNNABLE
  at weblogic.jdbc.wrapper.ResultSet.internalClose(ResultSet.java:178)
  at weblogic.jdbc.wrapper.Statement.closeAllResultSets(Statement.java:286)
  at weblogic.jdbc.wrapper.Statement.internalClose(Statement.java:395)
  at weblogic.jdbc.wrapper.Statement.internalClose(Statement.java:367)
  at weblogic.jdbc.wrapper.XAConnection.closeAllStatements(XAConnection.java:393)
  at weblogic.jdbc.wrapper.XAConnection.cleanup(XAConnection.java:406)
  at weblogic.jdbc.wrapper.XAConnection.releaseToPool(XAConnection.java:432)
  at weblogic.jdbc.jta.DataSource.removeTxAssoc(DataSource.java:1907)
  at weblogic.jdbc.jta.DataSource.prepare(DataSource.java:1090)
  at weblogic.transaction.internal.XAServerResourceInfo.prepare(XAServerResourceInfo.java:1408)
  at weblogic.transaction.internal.XAServerResourceInfo.prepare(XAServerResourceInfo.java:522)
  at weblogic.transaction.internal.ServerSCInfo.startPrepare(ServerSCInfo.java:411)
  at weblogic.transaction.internal.ServerTransactionImpl.localPrepare(ServerTransactionImpl.java:2709)
  at weblogic.transaction.internal.ServerTransactionImpl.globalPrepare(ServerTransactionImpl.java:2340)
  at weblogic.transaction.internal.ServerTransactionImpl.internalCommit(ServerTransactionImpl.java:300)
  at weblogic.transaction.internal.ServerTransactionImpl.commit(ServerTransactionImpl.java:260)
  at org.glassfish.transaction.TransactionManagerImplCommon.commit(TransactionManagerImplCommon.java:571)
  at org.springframework.transaction.jta.JtaTransactionManager.doCommit(JtaTransactionManager.java:1021)
  at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:761)
  at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:730)
  at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:150)
  at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:271)
  at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:77)
  at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:368)
  at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
  at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
  at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257)
  at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:198)
  at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
  at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:64)
  at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:67)
  at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:165)
  at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:144)
  at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:134)
  at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:304)
  at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:135)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:745)

Answer 1

您应该能够在 JTA 托管步骤中使用光标阅读器。在我正在从事的项目中，我们正是这样做的。我们使用 Atomikos 作为 XA TM。

这是我们使用的 XA/JTA 配置。也许它对你有一些用处：

@Bean(initMethod = "init", destroyMethod = "shutdownForce")
public UserTransactionService userTransactionService() {
    return new UserTransactionServiceImp(userTransactionServiceProperties());
}

@Bean(initMethod = "init", destroyMethod = "close")
@DependsOn("userTransactionService")
public UserTransactionManager atomikosTransactionManager() {
    UserTransactionManager userTransactionManager = new UserTransactionManager();
    userTransactionManager.setForceShutdown(true);
    userTransactionManager.setStartupTransactionService(false);
    return userTransactionManager;
}

@Bean
@DependsOn("userTransactionService")
public UserTransaction atomikosUserTransaction() throws SystemException {
    return new UserTransactionImp();
}

@Bean
@DependsOn("userTransactionService")
public JtaTransactionManager transactionManager() throws SystemException {
    JtaTransactionManager jtaTransactionManager = new JtaTransactionManager();
    jtaTransactionManager.setTransactionManager(atomikosTransactionManager());
    jtaTransactionManager.setUserTransaction(atomikosUserTransaction());
    jtaTransactionManager.setAllowCustomIsolationLevels(true);
    return jtaTransactionManager;
} 

我们所有的数据源都被实例化为 org.springframework.boot.jta.atomikos.AtomikosDataSourceBean。例如，一个 Ora-datasource 是这样实例化的：

    AtomikosDataSourceBean oraXaDs = new AtomikosDataSourceBean();
    oraXaDs.setXaDataSourceClassName(oraDsProp.getDatasourceClass());
    oraXaDs.setUniqueResourceName(oraDsProp.getInstancename());
    oraXaDs.setMinPoolSize(oraDsProp.getPoolMinSize());
    oraXaDs.setMaxPoolSize(oraDsProp.getPoolMaxSize());
    oraXaDs.setTestQuery(oraDsProp.getValidConnectionSQL());

    Properties oraXaDsProps = oraXaDs.getXaProperties();
    oraXaDsProps.setProperty("user", oraDsProp.getUser());
    oraXaDsProps.setProperty("password", oraDsProp.getPassword());
    oraXaDsProps.setProperty("URL", oraDsProp.getUrl());

Answer 2

我在这个问题上的两分钱：

首先是一些见解：

从数据库游标读取意味着打开一个连接，针对它触发一条 SQL 语句，并在整个批处理作业期间不断地读取行。这是有道理的，因为一个作业的输入数据通常可以用一条 SQL 语句来表征，但是执行它并预先从 ResultSet 中读取所有数据当然是没有解决方案的。我们在这里经常阅读只有一个问题：提交事务会关闭连接。 那么我们如何让它保持开放？简单的解决方案：它不参与事务。 Spring Batch 的 JdbcCursorItemReader 使用单独的连接打开游标，从而绕过由事务管理器管理的事务。 在应用服务器环境中，我们必须做更多工作才能使其工作。通常我们从应用服务器管理的 DataSource 获取连接，所有这些连接默认都参与事务。我们需要设置一个单独的不参与事务的数据源，并且只将它注入到基于游标的阅读器中。将它们注入其他任何地方可能会对交易安全造成很大损害。

您的问题基本上出在您的步骤中：（根据我可以在不查看您的数据源 xml 文件的情况下得出的结论 :)）

Step、Spring Batch Job Repository 和业务存储库（使用各种数据源）都使用 JTA 事务管理器。

Spring 提供的 JTA 事务管理器应该以 weblogic 处理您的 JTA 事务的方式使用。您只需将数据源（也应该在容器下）配置为使用 XA 感知驱动程序，并在您的应用程序，使用相应的事务管理器，并通过 JNDI 查找 XA 感知数据源

但是，如果您不能依赖容器（例如，您正在编写一个独立的应用程序等），您将需要找到一个有能力的事务管理器。 Atomikos 是最著名的免费 JTA/XA 库之一。

话虽如此，在您使用 JNDI 方式或 Atomikos 方式进行配置后，以下是使用多个数据源时应牢记的配置：

<batch:tasklet>
  <batch:transaction-attributes isolation="READ_COMMITTED" propagation="REQUIRES_NEW" timeout="200"/>
  <batch:chunk reader="myItemReader" writer="myItemWriter" commit-interval="20"/>
</batch:tasklet>

希望这能澄清这个问题。