如何从多个应用程序实例同时运行相同的弹簧批处理作业

Question

我创建了一个 spring 批处理作业，它在定义的时间使用调度程序执行。我们有同一个应用程序的多个实例，所以同一个作业会从多个实例同时执行。所有应用程序都使用相同的数据库。因此，我遇到了错误

堆栈跟踪：org.springframework.transaction.TransactionSystemException：无法提交 JDBC 事务；嵌套异常是 org.postgresql.util.PSQLException：错误：由于事务之间的读/写依赖关系而无法序列化访问详细信息：原因代码：在提交尝试期间在识别为枢轴时取消。提示：如果重试，事务可能会成功。在 org.springframework.jdbc.datasource.DataSourceTransactionManager.doCommit(DataSourceTransactionManager.java:335) 在 org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:743) 在 org.springframework.transaction.support.AbstractPlatformTransactionManager.commit (AbstractPlatformTransactionManager.java:711) 在 org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:385) 在 org.springframework.transactionAfterReturning(TransactionAspectSupport.java:631) 在 org.springframework.invokeWithinTransaction(TransactionAspectSupport.java:385)。 transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:118) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) at org.springframework.batch.core.repository.support.AbstractJobRepositoryFactoryBean$1.invoke (抽象JobRepositoryFactoryBean。 java:181) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212) at com.sun.proxy.$ org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:137) 上的 Proxy209.createJobExecution(Unknown Source)

我已经使用下面的代码说明了工作-

JobParameters params = new JobParametersBuilder()
            .addString("JobID", UUID.randomUUID().toString(), true)
            .toJobParameters();
jobLauncher.run(job, params);

并在 yaml 文件中进行以下配置

batch:
job:
  enabled: false
initialize-schema: always
repository:
  isolationlevelforcreate: ISOLATION_READ_COMMITTED

我添加了侦听器来检查任何正在运行的作业是否处于 STARTED 状态，如果找到任何正在运行的作业，则当前作业会停止

public class SingleInstanceListener implements JobExecutionListener {
@Autowired
private JobExplorer explorer;

@Override
public void beforeJob(JobExecution jobExecution) {
    String jobName = jobExecution.getJobInstance().getJobName();
    Set<JobExecution> executions = explorer.findRunningJobExecutions(jobName);
    if(executions.size() > 1) {
        jobExecution.stop();
    }
}

@Override
public void afterJob(JobExecution jobExecution) {

}

}

但我仍然遇到错误。如果只有一个应用程序实例，则作业运行良好。有没有办法从多个应用程序实例同时运行相同的作业，或者只允许一个作业从任何一个应用程序运行。

Answer 1

隔离级别未按照您对该属性的预期考虑。要自定义隔离级别，您需要通过BatchConfigurer 提供JobRepository。这是一个简单的例子：

@Configuration
@EnableBatchProcessing
public class MyJobConfig extends BasicBatchConfigurer {

    @Override
    protected JobRepository createJobRepository() throws Exception {
        JobRepositoryFactoryBean repositoryFactoryBean = new JobRepositoryFactoryBean();
        repositoryFactoryBean.setIsolationLevelForCreate("ISOLATION_READ_COMMITTED");
        // set other properties on the factory bean
        repositoryFactoryBean.afterPropertiesSet();
        return repositoryFactoryBean.getObject();
    }
}

这在此处的文档中进行了解释：Transaction Configuration for the JobRepository。

注意：BasicBatchConfigurer 来自 Spring Boot。如果你在没有 Spring Boot 的情况下使用 Spring Batch，你可以改为扩展 DefaultBatchConfigurer。