创建自定义 ItemReader 和 ItemWriter
到目前为止，本章讨论了 Spring Batch 中读取和写入的基本契约以及一些常见的实现方式。然而，这些实现都比较通用，可能存在许多开箱即用的实现无法覆盖的场景。本节将通过一个简单的示例，展示如何创建自定义的ItemReader和ItemWriter实现，并正确实现它们的契约。ItemReader还实现了ItemStream，以说明如何使读取器或写入器支持重启。
自定义ItemReader例举
为了本示例的目的，我们创建一个简单的 ItemReader 实现，该实现从提供的列表中读取数据。我们首先实现 ItemReader 的最基本契约，即 read 方法，如下面的代码所示：
public class CustomItemReader<T> implements ItemReader<T> {
List<T> items;
public CustomItemReader(List<T> items) {
this.items = items;
}
public T read() throws Exception, UnexpectedInputException,
NonTransientResourceException, ParseException {
if (!items.isEmpty()) {
return items.remove(0);
}
return null;
}
}上述类接收一个项目列表，并逐个返回这些项目，同时从列表中移除每个已返回的项目。当列表为空时，它返回 null，从而满足 ItemReader 的最基本要求，如下面的测试代码所示：
List<String> items = new ArrayList<>();
items.add("1");
items.add("2");
items.add("3");
ItemReader itemReader = new CustomItemReader<>(items);
assertEquals("1", itemReader.read());
assertEquals("2", itemReader.read());
assertEquals("3", itemReader.read());
assertNull(itemReader.read());让ItemReader可重启的
最后的挑战是使 ItemReader 可重启。目前，如果处理被中断并重新开始，ItemReader 必须从头开始。这在许多场景中实际上是有效的，但有时更希望批处理作业从中断处继续执行。关键的区分因素通常是读取器是有状态的还是无状态的。无状态读取器无需担心可重启性，但有状态读取器必须在重启时尝试恢复其最后已知的状态。因此，我们建议您尽可能保持自定义读取器为无状态，这样就不必担心可重启性问题。
如果您确实需要存储状态，则应使用 ItemStream 接口：
public class CustomItemReader<T> implements ItemReader<T>, ItemStream {
List<T> items;
int currentIndex = 0;
private static final String CURRENT_INDEX = "current.index";
public CustomItemReader(List<T> items) {
this.items = items;
}
public T read() throws Exception, UnexpectedInputException,
ParseException, NonTransientResourceException {
if (currentIndex < items.size()) {
return items.get(currentIndex++);
}
return null;
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
if (executionContext.containsKey(CURRENT_INDEX)) {
currentIndex = new Long(executionContext.getLong(CURRENT_INDEX)).intValue();
}
else {
currentIndex = 0;
}
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putLong(CURRENT_INDEX, new Long(currentIndex).longValue());
}
public void close() throws ItemStreamException {}
}每次调用 ItemStream update 方法时，ItemReader 的当前索引会存储在提供的 ExecutionContext 中，键为 'current.index'。当调用 ItemStream open 方法时，会检查 ExecutionContext 是否包含具有该键的条目。如果找到该键，则当前索引将移动到该位置。这是一个相当简单的示例，但它仍然满足通用约定：
ExecutionContext executionContext = new ExecutionContext();
((ItemStream)itemReader).open(executionContext);
assertEquals("1", itemReader.read());
((ItemStream)itemReader).update(executionContext);
List<String> items = new ArrayList<>();
items.add("1");
items.add("2");
items.add("3");
itemReader = new CustomItemReader<>(items);
((ItemStream)itemReader).open(executionContext);
assertEquals("2", itemReader.read());大多数 ItemReaders 拥有更复杂的重启逻辑。例如，JdbcCursorItemReader 会在游标中存储最后处理行的行 ID。
同样值得注意的是，ExecutionContext 中使用的键不应过于简单。这是因为同一个 ExecutionContext 会被用于 Step 中的所有 ItemStreams。在大多数情况下，只需在键前加上类名就足以保证唯一性。然而，在极少数情况下，如果在同一步骤中使用了两个相同类型的 ItemStream（例如需要两个文件作为输出时），则需要一个更唯一的名称。因此，许多 Spring Batch 的 ItemReader 和 ItemWriter 实现都提供了一个 setName() 属性，允许覆盖此键名。
自定义ItemWriter例举
实现自定义的 ItemWriter 在许多方面与上面的 ItemReader 示例类似，但在足够多的方面存在差异，因此值得单独作为一个示例。然而，添加可重启性本质上是一样的，因此本示例未涵盖此内容。与 ItemReader 示例一样，为了尽可能保持示例简单，这里使用了 List：
public class CustomItemWriter<T> implements ItemWriter<T> {
List<T> output = TransactionAwareProxyFactory.createTransactionalList();
public void write(Chunk<? extends T> items) throws Exception {
output.addAll(items);
}
public List<T> getOutput() {
return output;
}
}让ItemWriter可重启的
为了使 ItemWriter 可重启，我们将遵循与 ItemReader 相同的流程，添加并实现 ItemStream 接口以同步执行上下文。在示例中，我们可能需要统计已处理的项目数量，并将其作为尾部记录添加。如果需要这样做，我们可以在 ItemWriter 中实现 ItemStream，以便在流重新打开时从执行上下文中重建计数器。
在许多实际场景中，自定义的 ItemWriters 也会委托给另一个可重启的写入器（例如，当写入文件时），或者它写入的是事务性资源，因此由于其无状态特性而无需具备可重启能力。当您拥有有状态的写入器时，可能应该确保同时实现 ItemStream 和 ItemWriter。此外请记住，写入器的客户端需要知晓 ItemStream，因此您可能需要在配置中将其注册为流。