如何解决spring批处理中的InvalidIsolationLevelException？

Question

我正在使用spring batch，在执行批处理作业时，我得到了一个InvalidIsolationLevelException。当我使用MapJobRepositoryFactoryBean时，该作业可以工作，但我需要在数据库中生成batch_*表。

应用程序已经设置了事务管理器。

@Configuration
@ComponentScan({"com..."})
public class BatchConfig {
    private static final Logger LOGGER = LoggerFactory.getLogger(BatchConfig.class);

    private static final String REPORTS_STEP_NAME = "reportStep";
    private static final String REPORTS_JOB_NAME = "reportsJob";
    private static final int REPORTS_STEP_CHUNK_SIZE = 20;
    private static final String MAX_END_DATE = "2999-12-31 00:00:00";
    private static final String MAX_END_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    private static final String END_DATE_PARAM = "endDate";
    private static final String REPORT_STATUS_PARAM = "reportSend";
    private static final YesNo NO = new YesNo("N");

    @Autowired
    private ApplicationConfig applicationConfig;

    @Autowired
    private ReportProcessor reportProcessor;

    @Autowired
    private ReportItemWriter reportItemWriter;

    @Bean
    public JobBuilderFactory jobBuilderFactory() throws Exception {
        JobBuilderFactory jobBuilderFactory = new JobBuilderFactory(jobRepository());
        return jobBuilderFactory;
    }

    @Bean
    public StepBuilderFactory stepBuilderFactory() throws Exception {
        StepBuilderFactory stepBuilderFactory = new StepBuilderFactory(jobRepository(), applicationConfig.transactionManager());
        return stepBuilderFactory;
    }

    @Bean
    public PlatformTransactionManager transactionManager() {
        JpaTransactionManager jpaTransactionManager = new JpaTransactionManager();
        jpaTransactionManager.setDataSource(dataSource());
        jpaTransactionManager.setEntityManagerFactory(applicationConfig.entityManagerFactory().getObject());
        return jpaTransactionManager;
    }

    @Bean
    public JobRegistry jobRegistry() {
        return new MapJobRegistry();
    }

    @Bean
    public JobExplorer jobExplorer() throws Exception {
        JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
        jobExplorerFactoryBean.setDataSource(dataSource());
        jobExplorerFactoryBean.afterPropertiesSet();
        return jobExplorerFactoryBean.getObject();
    }

    @Bean
    public JobOperator jobOperator() throws Exception {
        SimpleJobOperator jobOperator = new SimpleJobOperator();
        jobOperator.setJobExplorer(jobExplorer());
        jobOperator.setJobRegistry(jobRegistry());
        jobOperator.setJobLauncher(jobLauncher());
        jobOperator.setJobRepository(jobRepository());
        return jobOperator;
    }

    @Bean
    public DataSource dataSource() {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("...");
        dataSource.setUrl("...");
        dataSource.setUsername("...");
        dataSource.setPassword("...");
        return dataSource;
    }

    @Bean
    public JobRepository jobRepository() throws Exception {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDataSource(dataSource());
        jobRepositoryFactoryBean.setTransactionManager(transactionManager());
        jobRepositoryFactoryBean.setIsolationLevelForCreate("ISOLATION_SERIALIZABLE");
        return jobRepositoryFactoryBean.getObject();
    }

    @Bean
    public JobLauncher jobLauncher() throws Exception {
        SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
        simpleJobLauncher.setJobRepository(jobRepository());
        return simpleJobLauncher;
    }

    public JpaPagingItemReader<FishingVessel> readFishingVesselRecords() throws ParseException {
        LOGGER.info("Inside JpaPagingItemReader");
        JpaPagingItemReader<FishingVessel> fishingVesselsReader = new JpaPagingItemReader<>();
        fishingVesselsReader.setEntityManagerFactory(applicationConfig.entityManagerFactory().getObject());
        fishingVesselsReader.setQueryString(FishingVessel.FISHING_VESSEL_REPORTS);
        fishingVesselsReader.setTransacted(false);
        fishingVesselsReader.setSaveState(false);
        Map<String, Object> params = createQueryParams();
        fishingVesselsReader.setParameterValues(params);
        fishingVesselsReader.setPageSize(100);
        return fishingVesselsReader;
    }

    private Map<String, Object> createQueryParams() throws ParseException {
        Map<String, Object> params = new HashMap<>();
        DateFormatter dateFormatter = new DateFormatter(MAX_END_DATE_FORMAT);
        Date date = dateFormatter.parse(MAX_END_DATE, Locale.ENGLISH);
        params.put(END_DATE_PARAM, date);
        //params.put(REPORT_STATUS_PARAM, NO);
        return params;
    }

    @Bean
    public Step reportsStep() throws Exception {
        DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
        attribute.setPropagationBehavior(Propagation.REQUIRED.value());
        attribute.setIsolationLevel(Isolation.SERIALIZABLE.value());
        attribute.setTimeout(30);

        return stepBuilderFactory()
                .get(REPORTS_STEP_NAME)
                .<FishingVessel, FishingVessel>chunk(REPORTS_STEP_CHUNK_SIZE)
                .reader(readFishingVesselRecords())
                .processor(reportProcessor)
                .writer(reportItemWriter)
                .transactionAttribute(attribute)
                .build();
    }

    @Bean
    public Job reportsJob() throws Exception {
        return jobBuilderFactory()
                .get(REPORTS_JOB_NAME)
                .start(reportsStep())
                .build();
    }
}

@Configuration
@ComponentScan({
    "com....components",
    "com....services",
    "com....converters",
    "com....controllers",
    "com....fluxFolder",
    "com....processors",
    "com....flux.service",
    "com....flux.gateway"
})
@EnableJpaRepositories("com....repositories")
@EnableTransactionManagement
@EnableJpaAuditing
@EnableWebMvc
@Import({SecurityConfig.class})
@PropertySource("classpath:application.properties")
public class ApplicationConfig {
    @Bean
    public LocalContainerEntityManagerFactoryBean entityManagerFactory() {

        LocalContainerEntityManagerFactoryBean factory = new LocalContainerEntityManagerFactoryBean();
        factory.setPersistenceXmlLocation("classpath:/META-INF/persistence.xml");
        return factory;
    }

    @Bean
    @Primary
    public PlatformTransactionManager transactionManager() {
        JpaTransactionManager transactionManager = new JpaTransactionManager();
        transactionManager.setEntityManagerFactory(entityManagerFactory().getObject());
        return transactionManager;
    }

    @Bean
    public static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() {
        return new PropertySourcesPlaceholderConfigurer();
    }

    @Bean
    public ViewResolver viewResolver() {
        UrlBasedViewResolver viewResolver = new UrlBasedViewResolver();
        viewResolver.setViewClass(JsfView.class);
        viewResolver.setSuffix(".xhtml");
        return viewResolver;
    }
}

@RestController
public class BatchController {
    private static final String JOB_ID = "JobID";

    private JobExecution jobExecution;

    @Autowired
    private JobOperator jobOperator;

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job reportsJob;

    @Autowired
    private JobExplorer jobExplorer;

    @GetMapping("/batch/enable")
    public String batchTest() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
        JobParameters jobParameters = new JobParametersBuilder()
                .addString(JOB_ID, String.valueOf(System.currentTimeMillis()))
                .toJobParameters();
        JobInstance job;
        jobExecution = new JobExecution(jobLauncher.run(reportsJob, jobParameters));
        return "batch job started";
    }

    @GetMapping("/batch/disable")
    public String disable() {
        String jobName = jobExecution.getJobInstance().getJobName();
        Set<JobExecution> jobExecutions = jobExplorer.findRunningJobExecutions(jobName);
        for (JobExecution je : jobExecutions) {
            try {
                jobOperator.stop(je.getId());
            } catch (NoSuchJobExecutionException e) {

            } catch (JobExecutionNotRunningException e) {

            }
        }
        return jobName + " disabled.";
    }
}

Answer 1

您可以将JobRepositoryFactoryBean中的隔离级别更改为默认：

 @Bean
public JobRepository jobRepository() throws Exception {
    JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
    jobRepositoryFactoryBean.setDataSource(dataSource());
    jobRepositoryFactoryBean.setTransactionManager(transactionManager());
    jobRepositoryFactoryBean.setIsolationLevelForCreate("ISOLATION_DEFAULT");
    return jobRepositoryFactoryBean.getObject();
}