Spring批处理：多线程步进错误配置

Question

以下是我的步骤：

@Bean
public Step autors(
    ItemReader<Autor> autorItemReader,
    AutorMappingItemProcessor processor,
    AutorPipeliningItemWriter unitatPipeliningWriter
) {
    return this.stepBuilderFactory
        .get("autors")
        .<Autor, AutorDenormalized>chunk(100)
        .reader(autorItemReader)
        .processor(processor)
        .writer(unitatPipeliningWriter)
        .build();
}

它工作得很好。

现在，我需要处理bt“chuck专用线程”。

我添加了以下配置：

@Bean
public Step autors(
    ItemReader<Autor> autorItemReader,
    AutorMappingItemProcessor processor,
    AutorPipeliningItemWriter unitatPipeliningWriter,
    TaskExecutor taskExecutor
) {
    return this.stepBuilderFactory
        .get("autors")
        .<Autor, AutorDenormalized>chunk(100)
        .reader(autorItemReader)
        .processor(processor)
        .writer(unitatPipeliningWriter)
        .taskExecutor(taskExecutor)
        .throttleLimit(4)
        .build();
}

@Bean
public TaskExecutor taskExecutor() {
    return new SimpleAsyncTaskExecutor("spring_batch");
}

问题出现在这里，因为我收到了这样的信息：

HikariPool-2 - Connection ConnectionID:1 ClientConnectionId: 8bee2b6e-d88e-4831-b9e5-163b52dca86c marked as broken because of SQLSTATE(08S01), ErrorCode(0)

com.microsoft.sqlserver.jdbc.SQLServerException: The connection is closed.

或者

HikariPool-2 - Connection ConnectionID:1 ClientConnectionId: 8bee2b6e-d88e-4831-b9e5-163b52dca86c marked as broken because of SQLSTATE(08S01), ErrorCode(0)

com.microsoft.sqlserver.jdbc.SQLServerException: The TDS protocol stream is not valid.

我相关的ItemReader是：

@Bean
public ItemReader<Autor> autorReader() {
    String sql = "select * from ...";

    JdbcCursorItemReader<Autor> jdbcCursorItemReader = new JdbcCursorItemReader<>();
    jdbcCursorItemReader.setDataSource(this.dataSource);
    jdbcCursorItemReader.setSql(sql);
    jdbcCursorItemReader.setVerifyCursorPosition(false);
    jdbcCursorItemReader.setRowMapper(this.autorMapper);

    return jdbcCursorItemReader;
}

我的数据源是：

@Bean
@JobDataSource
@ConfigurationProperties(prefix = "spring.job-datasource")
public DataSource jobDataSource() {
    return DataSourceBuilder.create().build();
}

和财产：

spring.job-datasource.jdbcUrl=jdbc:sqlserver://localhost;databaseName=ac_img_p
spring.job-datasource.username=sa
spring.job-datasource.password=StR0nGp4ss.
spring.job-datasource.driverClassName=com.microsoft.sqlserver.jdbc.SQLServerDriver
spring.job-datasource.initialization-mode=always

有什么想法吗？

Answer 1

JdbcCursorItemReader不是线程安全的，因为它扩展了不是线程安全的AbstractItemCounting ItemStreamItemReader。因此在多线程步骤中使用它是不正确的。您可以做的是用SynChronizedItemStreamReader装饰它：

@Bean
public SynchronizedItemStreamReader<Autor> autorReader() {
   String sql = "select * from ...";

   JdbcCursorItemReader<Autor> jdbcCursorItemReader = new JdbcCursorItemReader<>();
   jdbcCursorItemReader.setDataSource(this.dataSource);
   jdbcCursorItemReader.setSql(sql);
   jdbcCursorItemReader.setVerifyCursorPosition(false);
   jdbcCursorItemReader.setRowMapper(this.autorMapper);

   SynchronizedItemStreamReader<Autor> synchronizedReader = new SynchronizedItemStreamReader<>();
   synchronizedReader.setDelegate(jdbcCursorItemReader)
   return synchronizedReader;
}

否则，您可以使用线程安全读取器，例如JdbcPagingItemReader。

作为补充说明，您的autoreader方法应该返回实际类型，或者至少返回ItemStreamReader