Ohhnews

分类导航

$ cd ..
Spring Blog原文

Spring Batch 6.1.0-RC1 和 6.0.6 现已发布

#spring batch#版本发布#批处理框架#性能优化#jdbc

我谨代表团队及所有贡献者，很高兴地宣布 Spring Batch 6.1.0-RC1 和 6.0.6 现已可从 Maven Central 获取！这些版本带来了许多新功能、增强功能和缺陷修复。完整变更列表请查看发布说明：6.1.0-RC1 和 6.0.6。

本篇博客将带你了解 6.1.0-RC1 中的主要变更：

  • 依赖升级
  • 用于注入作业参数的 @JobParameter 注解
  • 新的 PageableItemReader
  • 与 JSON 库无关的 JsonLineMapper
  • 新的 JsonLineAggregator
  • 新的 RemoteChunkingManagerBuilder
  • 新的 DefaultChunkProcessor
  • JDBC DAO 迁移至 JdbcClient
  • 性能改进

依赖升级

Spring 依赖已升级至以下版本：

  • Spring Framework 7.1.0-RC1
  • Spring Integration 7.2.0-RC1
  • Spring Data 2026.1.0-RC1
  • Spring LDAP 4.2.0-RC1
  • Spring AMQP 4.2.0-RC1
  • Spring Kafka 4.2.0-RC1
  • Micrometer 1.18.0-RC1

用于注入作业参数的 @JobParameter 注解

过去，要将作业参数注入作业作用域或步骤作用域的 Bean，需要使用 SpEL 表达式；该表达式仅在运行时求值，很容易出错。新的 @JobParameter 注解是一种更简短的替代方案，它默认将作业参数名称设为被注解字段或参数的名称：

$ java
@Bean
@StepScope
public FlatFileItemReader<Foo> fooReader(@JobParameter String inputFile) {
    return new FlatFileItemReaderBuilder<Foo>()
            .resource(new FileSystemResource(inputFile))
            // other properties omitted
            .build();
}

它取代了 @Value("#{jobParameters['inputFile']}")。也可以显式指定名称，例如 @JobParameter("input.file.name")。与它所取代的表达式一样，该注解适用于在作业作用域或步骤作用域中定义的 Bean。当在方法或构造器参数上省略名称时，代码必须使用 -parameters 标志编译，这也是 Spring Boot Maven 和 Gradle 插件的默认设置。更多详情请参见 issue #5563。

新的 PageableItemReader

PageableItemReader 是 RepositoryItemReader 的类型安全替代方案。它不再通过名称反射调用 Repository 方法，而是接收一个从 Spring Data Pageable 到 Slice 的函数，该函数会在编译时检查：

$ java
PageableItemReader<Customer> reader = new PageableItemReaderBuilder<Customer>()
        .query(customerRepository::findAll)
        .sort(Sort.by(Sort.Direction.ASC, "id"))
        .pageSize(100)
        .name("customerReader")
        .build();

排序也可以使用 Spring Data 的类型安全属性引用来表达，例如 Sort.by(Sort.Direction.ASC, Customer::getId)，从而消除基于字符串的属性名。

与 JSON 库无关的 JsonLineMapper

过去，JsonLineMapper 只能将 JSON Lines 文件的每一行映射为 Map<String, Object>，而且仅支持 Jackson。现在它已改为泛型，并将行的转换委托给新的 JsonObjectUnmarshaller，从而不再依赖特定 JSON 库，并且能够将行映射为任意类型，例如领域类或 record。提供了基于 Jackson 和 Gson 的实现：

$ java
record Trade(String isin, int quantity, double price, String customer) {
}

FlatFileItemReader<Trade> reader = new FlatFileItemReaderBuilder<Trade>()
        .name("tradeReader")
        .resource(resource)
        .lineMapper(new JsonLineMapper<>(new JacksonJsonObjectUnmarshaller<>(Trade.class)))
        // or .lineMapper(new JsonLineMapper<>(new GsonJsonObjectUnmarshaller<>(Trade.class)))
        .build();

无参构造器以及接收 JsonMapper 的构造器已废弃，建议改用 JsonLineMapper(JsonObjectUnmarshaller)，并计划在 7.0 版本中移除。对于使用原始 JsonLineMapper 类型并直接调用其 mapLine 的代码来说，这是源代码和二进制不兼容变更。若代码只是将 mapper 传给 reader，则除重新编译外无需其他改动。更多详情请参见参考文档。

新的 JsonLineAggregator

JsonLineAggregator 是 JsonLineMapper 在写入端的对应实现：它是一个 LineAggregator，可使用 FlatFileItemWriter 将条目写入 JSON Lines 文件。与 mapper 一样，它不依赖特定 JSON 库，因为它将条目转换委托给 JsonObjectMarshaller，因此既可以使用 Jackson 实现，也可以使用 Gson 实现：

$ java
FlatFileItemWriter<Trade> writer = new FlatFileItemWriterBuilder<Trade>()
        .name("tradeWriter")
        .resource(new FileSystemResource("trades.jsonl"))
        .lineAggregator(new JsonLineAggregator<>(new JacksonJsonObjectMarshaller<>()))
        .build();

新的 RemoteChunkingManagerBuilder

RemoteChunkingManagerBuilder 是 RemoteChunkingWorkerBuilder 的对应实现，用于配置远程分块方案中的 manager 步骤。它只保存远程分块特有的设置（通道、消息模板、最大等待超时次数和限流阈值），而其他设置则通过 chunk(int) 返回标准的 ChunkOrientedStepBuilder 来配置。生成的步骤是 ChunkOrientedStep，并且该构建器不暴露任何已废弃的 API：

$ java
@Bean
public Step managerStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    return new RemoteChunkingManagerBuilder("managerStep", jobRepository)
            .outputChannel(requests())
            .inputChannel(replies())
            .chunk(100)
            .reader(itemReader())
            .transactionManager(transactionManager)
            .build();
}

RemoteChunkingManagerStepBuilder 和 RemoteChunkingManagerStepBuilderFactory 已废弃，建议改用此构建器，并计划在 7.0 版本中移除。请注意，事务管理器默认使用不依赖资源的事务管理器，而工厂此前使用应用上下文中的事务管理器，因此如果 manager 步骤的 item reader 需要事务性资源，就应显式设置事务管理器。

新的 DefaultChunkProcessor

SimpleChunkProcessor 自 6.0 起作为旧版面向分块处理引擎的一部分已被废弃，但它仍是在步骤之外处理分块的方式，例如在远程分块 worker 中或配合 ChunkTaskExecutorItemWriter 使用时。DefaultChunkProcessor 是这些用例的替代方案。它是一个最小化的 ChunkProcessor，会对分块中的每个条目应用可选的 ItemProcessor，并使用 ItemWriter 写入结果，例如 new DefaultChunkProcessor<>(itemProcessor, itemWriter)。与 SimpleChunkProcessor 不同，它不支持监听器，也不记录指标。RemoteChunkingWorkerBuilder 现在在内部使用它。

JDBC DAO 迁移至 JdbcClient

作业仓库 DAO 的 JDBC 实现（JdbcJobInstanceDao、JdbcJobExecutionDao、JdbcStepExecutionDao 和 JdbcExecutionContextDao）现在使用 Spring Framework 的 JdbcClient 而不是 JdbcOperations 来访问批处理元数据表。这使它们能够使用命名参数，并通过统一的流式 API 执行查询、更新和批量更新。该客户端是必需的协作者，因此现在通过构造器传入：

$ java
JdbcJobInstanceDao jobInstanceDao = new JdbcJobInstanceDao(JdbcClient.create(dataSource));

通过 @EnableBatchProcessing、@EnableJdbcJobRepository、JdbcDefaultBatchConfiguration、JdbcJobRepositoryFactoryBean 或 JobExplorerFactoryBean 配置作业仓库的应用不受影响，因为这些组件会自行构建 DAO。然而，对于直接实例化这些 DAO 的代码来说，这是源代码和二进制不兼容变更，因为它们的无参构造器已被移除。相应地，AbstractJdbcBatchMetadataDao#setJdbcTemplate(JdbcOperations) 和 #getJdbcTemplate() 已废弃，建议改用新的构造器和 #getJdbcClient()。更多详情请参见 issue #4804。

性能改进

此版本使作业仓库和分区步骤的多个部分速度更快、内存占用更低。以下是最主要的改进。

作业启动与重启

过去，重启作业会读取其作业实例的完整历史，加载每一次过往执行及其所有步骤。因此，作业重启越频繁，每次重启所需的 SQL 语句就越多：在 6.0.4 中，同一个失败作业的第 12 次重启发出了 130 条语句，每多一次过往执行就多 8 条。现在只检查最后一次执行，因此重启成本不再取决于历史记录。此改进在 6.0.6 中也可用。

查找作业实例的执行记录也更快了。JdbcJobExecutionDao#findJobExecutions 过去会为每次执行额外运行 3 个查询（N 次执行需要 1 + 3N 个查询）。现在无论 N 是多少，都只需要 2 个查询。对于有 10 次执行的作业实例，查询次数从 31 次降到 2 次，约为原来的 1/15。getLastJobExecution 从 4 个查询降到 2 个查询，减少了一半。

分区步骤

MessageChannelPartitionHandler 会定期向作业仓库询问其 worker 是否已完成。过去，它每次都会加载每个 worker 的步骤执行，而每个步骤执行都会连带加载整个作业执行及其所有步骤执行。在分区很多时，这会迅速累积：在一个已报告的案例中，一个包含 837 个分区的步骤在内存中保留了 837 × 837 = 700,569 个 StepExecution 对象，而轮询 worker 的节点使用了 15 GB 内存，是另一个执行相同工作量的节点（1.5 GB）的 10 倍。

现在，该处理器使用新的 JobRepository#countRunningStepExecutions(Collection<Long>) 仅统计正在运行的 worker，并在所有 worker 完成后只加载一次作业执行。在同一案例中，对象数量从 700,569 个减少到 837 个，约为原来的 1/837，并消除了轮询节点的额外内存占用：内存占用约为原来的 1/10。这些新方法是 default 方法，因此现有的 JobRepository 实现仍然可用；如果实现不支持计数，处理器会回退到之前的行为。

JDBC 作业仓库

JDBC DAO 中还移除了更多查询。每项优化在适用时都会节省一个查询：

  • 加载作业执行时，不再单独运行查询来查找其作业实例 ID，因为该 ID 已存在于 JOB_EXECUTION 表中。
  • 更新作业执行时，不再先运行 SELECT COUNT(*) 来检查行是否存在，因为更新操作本身就会报告行缺失。
  • JdbcJobExecutionDao#synchronizeStatus(StepExecution) 会节省一个查询。
  • 因并发变更导致作业执行或步骤执行更新失败时，不再额外运行一个查询，仅为了在错误消息中显示当前版本。

最后两项改进在 6.0.6 中也可用。

反馈

我要感谢所有为本次发布做出贡献的人！随着 6.1 GA 版本日益临近，我们期待在 Github Issues、Github Discussions 和 X 上收到您的反馈。

Spring Batch 主页|GitHub 上的源码|参考文档