常见批处理模式

一些批处理 Job 可以完全由 Spring Batch 中的现成组件组装而成。例如，可以配置 ItemReader 和 ItemWriter 的实现来覆盖广泛的场景。然而，在大多数情况下，必须编写自定义代码。应用程序开发人员的主要 API 入口点是 Tasklet、ItemReader、ItemWriter 和各种监听器接口。大多数简单的批处理 Job 可以使用 Spring Batch ItemReader 提供的现成输入，但通常在处理和写入过程中存在需要开发人员实现 ItemWriter 或 ItemProcessor 的自定义需求。

在本章中，我们提供了一些自定义业务逻辑中常见模式的示例。这些示例主要涉及监听器接口。需要注意的是，如果合适，ItemReader 或 ItemWriter 也可以实现监听器接口。

记录 Item 处理和失败

一个常见的用例是需要在 Step 中逐个 Item 对错误进行特殊处理，例如记录到特殊通道或将记录插入数据库。面向块的 Step（由 Step 工厂 bean 创建）允许用户通过简单的 ItemReadListener 来处理 read 错误，以及通过 ItemWriteListener 来处理 write 错误来实现此用例。以下代码片段展示了一个同时记录读写失败的监听器

public class ItemFailureLoggerListener extends ItemListenerSupport {

    private static Log logger = LogFactory.getLog("item.error");

    public void onReadError(Exception ex) {
        logger.error("Encountered error on read", e);
    }

    public void onWriteError(Exception ex, List<? extends Object> items) {
        logger.error("Encountered error on write", ex);
    }
}

实现此监听器后，必须将其注册到 Step。

  • Java

  • XML

以下示例展示了如何在 Java 中将监听器注册到 Step

Java 配置
@Bean
public Step simpleStep(JobRepository jobRepository) {
	return new StepBuilder("simpleStep", jobRepository)
				...
				.listener(new ItemFailureLoggerListener())
				.build();
}

以下示例展示了如何在 XML 中将监听器注册到 Step

XML 配置
<step id="simpleStep">
...
<listeners>
    <listener>
        <bean class="org.example...ItemFailureLoggerListener"/>
    </listener>
</listeners>
</step>
如果你的监听器在 onError() 方法中执行任何操作，它必须位于将要回滚的事务内部。如果你需要在 onError() 方法中使用事务资源（例如数据库），考虑为该方法添加声明式事务（详见 Spring Core 参考指南），并将其传播属性设置为 REQUIRES_NEW。

出于业务原因手动停止 Job

Spring Batch 通过 JobOperator 接口提供了一个 stop() 方法，但这实际上是供操作员使用，而不是应用程序员使用。有时，从业务逻辑内部停止 Job 执行更方便或更有意义。

最简单的方法是抛出 RuntimeException（既不会无限重试也不会被跳过）。例如，可以使用自定义异常类型，如下例所示

public class PoisonPillItemProcessor<T> implements ItemProcessor<T, T> {

    @Override
    public T process(T item) throws Exception {
        if (isPoisonPill(item)) {
            throw new PoisonPillException("Poison pill detected: " + item);
        }
        return item;
    }
}

停止 Step 执行的另一种简单方法是从 ItemReader 返回 null，如下例所示

public class EarlyCompletionItemReader implements ItemReader<T> {

    private ItemReader<T> delegate;

    public void setDelegate(ItemReader<T> delegate) { ... }

    public T read() throws Exception {
        T item = delegate.read();
        if (isEndItem(item)) {
            return null; // end the step here
        }
        return item;
    }

}

前面的示例实际上依赖于一个事实，即存在 CompletionPolicy 策略的默认实现，当要处理的 Item 为 null 时，该实现会发出批处理完成的信号。可以通过 SimpleStepFactoryBean 实现更复杂的完成策略并将其注入到 Step 中。

  • Java

  • XML

以下示例展示了如何在 Java 中将完成策略注入到 Step 中

Java 配置
@Bean
public Step simpleStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("simpleStep", jobRepository)
				.<String, String>chunk(new SpecialCompletionPolicy(), transactionManager)
				.reader(reader())
				.writer(writer())
				.build();
}

以下示例展示了如何在 XML 中将完成策略注入到 Step 中

XML 配置
<step id="simpleStep">
    <tasklet>
        <chunk reader="reader" writer="writer" commit-interval="10"
               chunk-completion-policy="completionPolicy"/>
    </tasklet>
</step>

<bean id="completionPolicy" class="org.example...SpecialCompletionPolicy"/>

另一种方法是在 StepExecution 中设置一个标志，框架中的 Step 实现会在 Item 处理之间检查此标志。为了实现此替代方法，我们需要访问当前的 StepExecution，这可以通过实现 StepListener 并将其注册到 Step 来实现。以下示例展示了一个设置标志的监听器

public class CustomItemWriter extends ItemListenerSupport implements StepListener {

    private StepExecution stepExecution;

    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    public void afterRead(Object item) {
        if (isPoisonPill(item)) {
            stepExecution.setTerminateOnly();
       }
    }

}

设置标志后，默认行为是 Step 抛出 JobInterruptedException。可以通过 StepInterruptionPolicy 控制此行为。但是，唯一的选择是抛出异常或不抛出异常，因此这总是 Job 的异常结束。

添加页脚记录

通常，在写入平面文件时，需要在文件末尾附加一个“页脚”记录，以便在所有处理完成后添加。这可以使用 Spring Batch 提供的 FlatFileFooterCallback 接口来实现。FlatFileFooterCallback（及其对应物 FlatFileHeaderCallback）是 FlatFileItemWriter 的可选属性，可以添加到 Item Writer 中。

  • Java

  • XML

以下示例展示了如何在 Java 中使用 FlatFileHeaderCallback 和 FlatFileFooterCallback

Java 配置
@Bean
public FlatFileItemWriter<String> itemWriter(Resource outputResource) {
	return new FlatFileItemWriterBuilder<String>()
			.name("itemWriter")
			.resource(outputResource)
			.lineAggregator(lineAggregator())
			.headerCallback(headerCallback())
			.footerCallback(footerCallback())
			.build();
}

以下示例展示了如何在 XML 中使用 FlatFileHeaderCallback 和 FlatFileFooterCallback

XML 配置
<bean id="itemWriter" class="org.spr...FlatFileItemWriter">
    <property name="resource" ref="outputResource" />
    <property name="lineAggregator" ref="lineAggregator"/>
    <property name="headerCallback" ref="headerCallback" />
    <property name="footerCallback" ref="footerCallback" />
</bean>

页脚回调接口只有一个方法，该方法在需要写入页脚时被调用，如以下接口定义所示

public interface FlatFileFooterCallback {

    void writeFooter(Writer writer) throws IOException;

}

编写摘要页脚

涉及页脚记录的一个常见需求是在输出过程中汇总信息，并将此信息附加到文件末尾。此页脚通常用作文件的摘要或提供校验和。

例如，如果一个批处理 Job 正在将 Trade 记录写入平面文件，并且要求将所有 Trade 的总金额放在页脚中，则可以使用以下 ItemWriter 实现

public class TradeItemWriter implements ItemWriter<Trade>,
                                        FlatFileFooterCallback {

    private ItemWriter<Trade> delegate;

    private BigDecimal totalAmount = BigDecimal.ZERO;

    public void write(Chunk<? extends Trade> items) throws Exception {
        BigDecimal chunkTotal = BigDecimal.ZERO;
        for (Trade trade : items) {
            chunkTotal = chunkTotal.add(trade.getAmount());
        }

        delegate.write(items);

        // After successfully writing all items
        totalAmount = totalAmount.add(chunkTotal);
    }

    public void writeFooter(Writer writer) throws IOException {
        writer.write("Total Amount Processed: " + totalAmount);
    }

    public void setDelegate(ItemWriter delegate) {...}
}

此 TradeItemWriter 存储一个 totalAmount 值，该值随写入的每个 Trade 项的 amount 增加。处理完最后一个 Trade 后，框架会调用 writeFooter，将 totalAmount 写入文件。请注意，write 方法使用了一个临时变量 chunkTotal，用于存储该块中 Trade 金额的总和。这样做是为了确保，如果在 write 方法中发生跳过，totalAmount 保持不变。只有在 write 方法结束时，一旦我们确定没有抛出异常，我们才会更新 totalAmount。

为了调用 writeFooter 方法，必须将 TradeItemWriter（它实现了 FlatFileFooterCallback）作为 footerCallback 注入到 FlatFileItemWriter 中。

  • Java

  • XML

以下示例展示了如何在 Java 中注入 TradeItemWriter

Java 配置
@Bean
public TradeItemWriter tradeItemWriter() {
	TradeItemWriter itemWriter = new TradeItemWriter();

	itemWriter.setDelegate(flatFileItemWriter(null));

	return itemWriter;
}

@Bean
public FlatFileItemWriter<String> flatFileItemWriter(Resource outputResource) {
	return new FlatFileItemWriterBuilder<String>()
			.name("itemWriter")
			.resource(outputResource)
			.lineAggregator(lineAggregator())
			.footerCallback(tradeItemWriter())
			.build();
}

以下示例展示了如何在 XML 中注入 TradeItemWriter

XML 配置
<bean id="tradeItemWriter" class="..TradeItemWriter">
    <property name="delegate" ref="flatFileItemWriter" />
</bean>

<bean id="flatFileItemWriter" class="org.spr...FlatFileItemWriter">
   <property name="resource" ref="outputResource" />
   <property name="lineAggregator" ref="lineAggregator"/>
   <property name="footerCallback" ref="tradeItemWriter" />
</bean>

目前为止 TradeItemWriter 的编写方式只有在 Step 不可重启的情况下才能正确工作。这是因为该类是有状态的（因为它存储了 totalAmount），但 totalAmount 没有持久化到数据库。因此，在重启时无法检索到它。为了使该类可重启，应实现 ItemStream 接口以及 open 和 update 方法，如下例所示

public void open(ExecutionContext executionContext) {
    if (executionContext.containsKey("total.amount") {
        totalAmount = (BigDecimal) executionContext.get("total.amount");
    }
}

public void update(ExecutionContext executionContext) {
    executionContext.put("total.amount", totalAmount);
}

update 方法在将 ExecutionContext 对象持久化到数据库之前，将 totalAmount 的最新版本存储到 ExecutionContext 中。open 方法从 ExecutionContext 中检索任何现有的 totalAmount，并将其用作处理的起点，从而允许 TradeItemWriter 在重启时从上次 Step 运行中断的地方继续执行。

基于驱动查询的 ItemReader

在关于 Reader 和 Writer 的章节中，讨论了使用分页的数据库输入。许多数据库厂商（例如 DB2）具有非常悲观的锁定策略，如果正在读取的表还需要被在线应用程序的其他部分使用，可能会导致问题。此外，在超大数据集上打开游标也可能导致某些厂商的数据库出现问题。因此，许多项目倾向于使用“驱动查询”方法来读取数据。这种方法通过迭代键而不是需要返回的整个对象来工作，如下图所示

Driving Query Job
图 1. 驱动查询 Job

如你所见，前面图像中显示的示例使用了与基于游标的示例相同的 'FOO' 表。但是，SQL 语句中选择的不是整行，而是仅选择 ID。因此，从 read 返回的不是 FOO 对象，而是 Integer。然后可以使用此数字查询“详细信息”，即完整的 Foo 对象，如下图所示

Driving Query Example
图 2. 驱动查询示例

应该使用 ItemProcessor 将从驱动查询中获得的键转换为完整的 Foo 对象。可以使用现有的 DAO 根据键查询完整的对象。

多行记录

虽然平面文件通常每条记录都限制在一行内，但文件包含跨越多行且具有多种格式的记录也很常见。以下文件摘录显示了此类安排的示例

HEA;0013100345;2007-02-15
NCU;Smith;Peter;;T;20014539;F
BAD;;Oak Street 31/A;;Small Town;00235;IL;US
FOT;2;2;267.34

以 'HEA' 开头的行和以 'FOT' 开头的行之间的所有内容被视为一条记录。为了正确处理这种情况，必须考虑以下几点

  • ItemReader 必须一次读取多行记录的所有行作为一个组，而不是一次读取一条记录，以便将其完整地传递给 ItemWriter。

  • 每种行类型可能需要以不同的方式进行标记化。

因为单条记录跨越多行，并且我们可能不知道有多少行，所以 ItemReader 必须小心，始终读取整条记录。为此，应将自定义 ItemReader 实现为 FlatFileItemReader 的包装器。

  • Java

  • XML

以下示例展示了如何在 Java 中实现自定义 ItemReader

Java 配置
@Bean
public MultiLineTradeItemReader itemReader() {
	MultiLineTradeItemReader itemReader = new MultiLineTradeItemReader();

	itemReader.setDelegate(flatFileItemReader());

	return itemReader;
}

@Bean
public FlatFileItemReader flatFileItemReader() {
	FlatFileItemReader<Trade> reader = new FlatFileItemReaderBuilder<>()
			.name("flatFileItemReader")
			.resource(new ClassPathResource("data/iosample/input/multiLine.txt"))
			.lineTokenizer(orderFileTokenizer())
			.fieldSetMapper(orderFieldSetMapper())
			.build();
	return reader;
}

以下示例展示了如何在 XML 中实现自定义 ItemReader

XML 配置
<bean id="itemReader" class="org.spr...MultiLineTradeItemReader">
    <property name="delegate">
        <bean class="org.springframework.batch.item.file.FlatFileItemReader">
            <property name="resource" value="data/iosample/input/multiLine.txt" />
            <property name="lineMapper">
                <bean class="org.spr...DefaultLineMapper">
                    <property name="lineTokenizer" ref="orderFileTokenizer"/>
                    <property name="fieldSetMapper" ref="orderFieldSetMapper"/>
                </bean>
            </property>
        </bean>
    </property>
</bean>

为确保每行都正确标记化（这对于固定长度输入尤其重要），可以在委托 FlatFileItemReader 上使用 PatternMatchingCompositeLineTokenizer。详见 Reader 和 Writer 章节中的 FlatFileItemReader。然后，委托 Reader 使用 PassThroughFieldSetMapper 将每行的 FieldSet 返回给包装器 ItemReader。

  • Java

  • XML

以下示例展示了如何在 Java 中确保每行都被正确标记化

Java 内容
@Bean
public PatternMatchingCompositeLineTokenizer orderFileTokenizer() {
	PatternMatchingCompositeLineTokenizer tokenizer =
			new PatternMatchingCompositeLineTokenizer();

	Map<String, LineTokenizer> tokenizers = new HashMap<>(4);

	tokenizers.put("HEA*", headerRecordTokenizer());
	tokenizers.put("FOT*", footerRecordTokenizer());
	tokenizers.put("NCU*", customerLineTokenizer());
	tokenizers.put("BAD*", billingAddressLineTokenizer());

	tokenizer.setTokenizers(tokenizers);

	return tokenizer;
}

以下示例展示了如何在 XML 中确保每行都被正确标记化

XML 内容
<bean id="orderFileTokenizer" class="org.spr...PatternMatchingCompositeLineTokenizer">
    <property name="tokenizers">
        <map>
            <entry key="HEA*" value-ref="headerRecordTokenizer" />
            <entry key="FOT*" value-ref="footerRecordTokenizer" />
            <entry key="NCU*" value-ref="customerLineTokenizer" />
            <entry key="BAD*" value-ref="billingAddressLineTokenizer" />
        </map>
    </property>
</bean>

此包装器必须能够识别记录的结尾，以便可以持续对其委托调用 read() 直到到达结尾。对于读取的每一行，包装器都应构建要返回的 Item。一旦到达页脚，就可以返回该 Item，将其交付给 ItemProcessor 和 ItemWriter，如下例所示

private FlatFileItemReader<FieldSet> delegate;

public Trade read() throws Exception {
    Trade t = null;

    for (FieldSet line = null; (line = this.delegate.read()) != null;) {
        String prefix = line.readString(0);
        if (prefix.equals("HEA")) {
            t = new Trade(); // Record must start with header
        }
        else if (prefix.equals("NCU")) {
            Assert.notNull(t, "No header was found.");
            t.setLast(line.readString(1));
            t.setFirst(line.readString(2));
            ...
        }
        else if (prefix.equals("BAD")) {
            Assert.notNull(t, "No header was found.");
            t.setCity(line.readString(4));
            t.setState(line.readString(6));
          ...
        }
        else if (prefix.equals("FOT")) {
            return t; // Record must end with footer
        }
    }
    Assert.isNull(t, "No 'END' was found.");
    return null;
}

执行系统命令

许多批处理 Job 需要从批处理 Job 内部调用外部命令。这样的过程可以由调度程序单独启动，但这会失去关于运行的通用元数据的优势。此外，多步骤的 Job 也需要拆分成多个 Job。

由于这种需求非常普遍，Spring Batch 提供了一个用于调用系统命令的 Tasklet 实现。

  • Java

  • XML

以下示例展示了如何在 Java 中调用外部命令

Java 配置
@Bean
public SystemCommandTasklet tasklet() {
	SystemCommandTasklet tasklet = new SystemCommandTasklet();

	tasklet.setCommand("echo hello");
	tasklet.setTimeout(5000);

	return tasklet;
}

以下示例展示了如何在 XML 中调用外部命令

XML 配置
<bean class="org.springframework.batch.core.step.tasklet.SystemCommandTasklet">
    <property name="command" value="echo hello" />
    <!-- 5 second timeout for the command to complete -->
    <property name="timeout" value="5000" />
</bean>

处理未找到输入时的 Step 完成

在许多批处理场景中，在数据库或文件中找不到要处理的行并非异常情况。Step 被简单地认为没有找到工作，并以读取 0 个 Item 完成。Spring Batch 中开箱即用提供的所有 ItemReader 实现都默认为此方法。如果在存在输入的情况下（通常是文件命名错误或发生类似问题时）没有输出任何内容，可能会导致一些困惑。因此，应检查元数据本身以确定框架发现了多少工作需要处理。但是，如果找不到输入被视为异常情况怎么办？在这种情况下，通过编程方式检查元数据以了解没有处理 Item 并导致失败是最佳解决方案。由于这是一个常见的用例，Spring Batch 提供了一个具有此功能的监听器，如 NoWorkFoundStepExecutionListener 的类定义所示

public class NoWorkFoundStepExecutionListener implements StepExecutionListener {

    public ExitStatus afterStep(StepExecution stepExecution) {
        if (stepExecution.getReadCount() == 0) {
            return ExitStatus.FAILED;
        }
        return null;
    }

}

前面的 StepExecutionListener 在 'afterStep' 阶段检查 StepExecution 的 readCount 属性，以确定是否未读取任何 Item。如果是这种情况，则返回退出码 FAILED，表明 Step 应该失败。否则，返回 null，这不会影响 Step 的状态。

将数据传递给后续 Step

通常将信息从一个 Step 传递到另一个 Step 是很有用的。这可以通过 ExecutionContext 来完成。需要注意的是，有两个 ExecutionContext：一个在 Step 级别，一个在 Job 级别。Step 的 ExecutionContext 仅在 Step 运行时保留，而 Job 的 ExecutionContext 贯穿整个 Job。另一方面，Step 的 ExecutionContext 在 Step 每次提交一个块时更新，而 Job 的 ExecutionContext 仅在每个 Step 结束时更新。

这种分离的结果是，在 Step 执行期间，所有数据都必须放在 Step 的 ExecutionContext 中。这样做可以确保数据在 Step 运行时正确存储。如果数据存储到 Job 的 ExecutionContext 中，则在 Step 执行期间不会持久化。如果 Step 失败，该数据将丢失。

public class SavingItemWriter implements ItemWriter<Object> {
    private StepExecution stepExecution;

    public void write(Chunk<? extends Object> items) throws Exception {
        // ...

        ExecutionContext stepContext = this.stepExecution.getExecutionContext();
        stepContext.put("someKey", someObject);
    }

    @BeforeStep
    public void saveStepExecution(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }
}

为了使数据对后续 Step 可用，必须在 Step 完成后将其“提升”到 Job 的 ExecutionContext。Spring Batch 为此目的提供了 ExecutionContextPromotionListener。必须使用 ExecutionContext 中需要提升的数据相关的键来配置监听器。它还可以选择配置一个退出码模式列表，在此模式下进行提升（默认值为 COMPLETED）。与所有监听器一样，它必须在 Step 上注册。

  • Java

  • XML

以下示例展示了如何在 Java 中将 Step 提升到 Job 的 ExecutionContext

Java 配置
@Bean
public Job job1(JobRepository jobRepository, Step step1, Step step2) {
	return new JobBuilder("job1", jobRepository)
				.start(step1)
				.next(step2)
				.build();
}

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
	return new StepBuilder("step1", jobRepository)
				.<String, String>chunk(10, transactionManager)
				.reader(reader())
				.writer(savingWriter())
				.listener(promotionListener())
				.build();
}

@Bean
public ExecutionContextPromotionListener promotionListener() {
	ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();

	listener.setKeys(new String[] {"someKey"});

	return listener;
}

以下示例展示了如何在 XML 中将 Step 提升到 Job 的 ExecutionContext

XML 配置
<job id="job1">
    <step id="step1">
        <tasklet>
            <chunk reader="reader" writer="savingWriter" commit-interval="10"/>
        </tasklet>
        <listeners>
            <listener ref="promotionListener"/>
        </listeners>
    </step>

    <step id="step2">
       ...
    </step>
</job>

<beans:bean id="promotionListener" class="org.spr....ExecutionContextPromotionListener">
    <beans:property name="keys">
        <list>
            <value>someKey</value>
        </list>
    </beans:property>
</beans:bean>

最后，必须从 Job 的 ExecutionContext 中检索保存的值，如下例所示

public class RetrievingItemWriter implements ItemWriter<Object> {
    private Object someObject;

    public void write(Chunk<? extends Object> items) throws Exception {
        // ...
    }

    @BeforeStep
    public void retrieveInterstepData(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        ExecutionContext jobContext = jobExecution.getExecutionContext();
        this.someObject = jobContext.get("someKey");
    }
}