Spring批处理：多线程步骤错误配置答案

【问题标题】：Spring batch: multithread step misconfigurationSpring批处理：多线程步骤错误配置
【发布时间】：2021-03-01 20:40:23
【问题描述】：

这是我的步骤：

@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.

我的相关@987654328@是：

@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

有什么想法吗？

【问题讨论】：

在提出新问题之前，请接受答案或就您以前的问题提供反馈：stackoverflow.com/questions/64888740/…、stackoverflow.com/questions/64871126/…。 SO 是一个协作平台，目标不是获得答案并继续前进。您需要接受/赞成/反对答案或至少提供反馈：stackoverflow.com/help/someone-answers。

标签： spring-batch hikaricp

【解决方案1】：

JdbcCursorItemReader 不是线程安全的，因为它扩展了AbstractItemCountingItemStreamItemReader，即not thread-safe。所以在多线程步骤中使用它是不正确的。你可以做的是用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 这样的线程安全阅读器。

附带说明一下，您的 autorReader 方法应返回实际类型或至少返回 ItemStreamReader<Autor>，以便 Spring Batch 正确代理您的阅读器。

【讨论】：