Spring Batch -项目处理

ItemReader 和 ItemWriter 接口​对于其特定 任务，但是如果要在编写之前插入业务逻辑怎么办？两者兼而有之的一个选项 读写是使用复合模式：创建一个包含 另一个或包含另一个 .以下 代码显示了一个示例：​​ItemWriter​​​​ItemWriter​​​​ItemReader​​​​ItemReader​​

public class CompositeItemWriter<T> implements ItemWriter<T> {

    ItemWriter<T> itemWriter;

    public CompositeItemWriter(ItemWriter<T> itemWriter) {
        this.itemWriter = itemWriter;
    }

    public void write(Chunk<? extends T> items) throws Exception {
        //Add business logic here
       itemWriter.write(items);
    }

    public void setDelegate(ItemWriter<T> itemWriter){
        this.itemWriter = itemWriter;
    }
}

前面的类包含另一个类，它在具有 提供了一些业务逻辑。此模式可以很容易地用于 好吧，也许是为了根据提供的输入获得更多参考数据 主要。如果您需要控制对自己的调用，它也很有用。 但是，如果您只想“转换”传入的用于写入的项目，则在它 实际写的，你不需要你自己。您可以只修改该项目。为此 场景，Spring Batch 提供了接口，如下所示 接口定义显示：​​ItemWriter​​​​ItemReader​​​​ItemReader​​​​write​​​​write​​​​ItemProcessor​​

public interface ItemProcessor<I, O> {

    O process(I item) throws Exception;
}

安很简单。给定一个对象，转换它并返回另一个对象。这 提供的对象可能属于同一类型，也可能不是同一类型。关键是业务逻辑可能 在流程中应用，完全由开发人员创建 逻辑。可以直接连接到一个步骤中。例如，假设 a 提供了一个类型类，并且在写出之前需要将其转换为类型。以下示例显示执行 转换：​​ItemProcessor​​​​ItemProcessor​​​​ItemReader​​​​Foo​​​​Bar​​​​ItemProcessor​​

public class Foo {}

public class Bar {
    public Bar(Foo foo) {}
}

public class FooProcessor implements ItemProcessor<Foo, Bar> {
    public Bar process(Foo foo) throws Exception {
        //Perform simple transformation, convert a Foo to a Bar
        return new Bar(foo);
    }
}

public class BarWriter implements ItemWriter<Bar> {
    public void write(Chunk<? extends Bar> bars) throws Exception {
        //write bars
    }
}

在前面的示例中，有一个名为 的类 、 和一个名为 的类 遵循接口的命名。转换是 很简单，但任何类型的转换都可以在这里完成。写入对象，如果提供了任何其他类型，则引发异常。同样，如果提供除 a 以外的任何内容，则会引发异常。然后可以将其注入到 ，如以下示例所示：​​Foo​​​​Bar​​​​FooProcessor​​​​ItemProcessor​​​​BarWriter​​​​Bar​​​​FooProcessor​​​​Foo​​​​FooProcessor​​​​Step​​

XML 配置

<job id="ioSampleJob">
    <step name="step1">
        <tasklet>
            <chunk reader="fooReader" processor="fooProcessor" writer="barWriter"
                   commit-interval="2"/>
        </tasklet>
    </step>
</job>

爪哇配置

@Bean
public Job ioSampleJob(JobRepository jobRepository) {
  return new JobBuilder("ioSampleJob", jobRepository)
        .start(step1())
        .build();
}

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
  return new StepBuilder("step1", jobRepository)
        .<Foo, Bar>chunk(2, transactionManager)
        .reader(fooReader())
        .processor(fooProcessor())
        .writer(barWriter())
        .build();
}

与 或 之间的区别是 对于 是可选的。​​ItemProcessor​​​​ItemReader​​​​ItemWriter​​​​ItemProcessor​​​​Step​​

链接项处理器

在许多情况下，执行单个转换很有用，但是如果需要，该怎么办 将多个实现“链接”在一起？您可以使用 前面提到的复合模式。要更新上一个，单个 转换示例转换为 ，转换为 并写出，如以下示例所示：​​ItemProcessor​​​​Foo​​​​Bar​​​​Foobar​​

public class Foo {}

public class Bar {
    public Bar(Foo foo) {}
}

public class Foobar {
    public Foobar(Bar bar) {}
}

public class FooProcessor implements ItemProcessor<Foo, Bar> {
    public Bar process(Foo foo) throws Exception {
        //Perform simple transformation, convert a Foo to a Bar
        return new Bar(foo);
    }
}

public class BarProcessor implements ItemProcessor<Bar, Foobar> {
    public Foobar process(Bar bar) throws Exception {
        return new Foobar(bar);
    }
}

public class FoobarWriter implements ItemWriter<Foobar>{
    public void write(Chunk<? extends Foobar> items) throws Exception {
        //write items
    }
}

A 和 a 可以“链接”在一起给出结果，如下面的例子所示：​​FooProcessor​​​​BarProcessor​​​​Foobar​​

CompositeItemProcessor<Foo,Foobar> compositeProcessor =
                                      new CompositeItemProcessor<Foo,Foobar>();
List itemProcessors = new ArrayList();
itemProcessors.add(new FooProcessor());
itemProcessors.add(new BarProcessor());
compositeProcessor.setDelegates(itemProcessors);

与前面的示例一样，您可以将复合处理器配置为：​​Step​​

XML 配置

<job id="ioSampleJob">
    <step name="step1">
        <tasklet>
            <chunk reader="fooReader" processor="compositeItemProcessor" writer="foobarWriter"
                   commit-interval="2"/>
        </tasklet>
    </step>
</job>

<bean id="compositeItemProcessor"
      class="org.springframework.batch.item.support.CompositeItemProcessor">
    <property name="delegates">
        <list>
            <bean class="..FooProcessor" />
            <bean class="..BarProcessor" />
        </list>
    </property>
</bean>

爪哇配置

@Bean
public Job ioSampleJob(JobRepository jobRepository) {
  return new JobBuilder("ioSampleJob", jobRepository)
        .start(step1())
        .build();
}

@Bean
public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
  return tnew StepBuilder("step1", jobRepository)
        .<Foo, Foobar>chunk(2, transactionManager)
        .reader(fooReader())
        .processor(compositeProcessor())
        .writer(foobarWriter())
        .build();
}

@Bean
public CompositeItemProcessor compositeProcessor() {
  List<ItemProcessor> delegates = new ArrayList<>(2);
  delegates.add(new FooProcessor());
  delegates.add(new BarProcessor());

  CompositeItemProcessor processor = new CompositeItemProcessor();

  processor.setDelegates(delegates);

  return processor;
}

筛选记录

项目处理器的一个典型用途是在记录传递到 这。筛选是一种不同于跳过的操作。跳过表示 记录无效，而筛选指示记录不应 写。​​ItemWriter​​

例如，考虑一个批处理作业，该作业读取包含三种不同类型的 记录：要插入的记录、要更新的记录和要删除的记录。如果记录删除 系统不支持，我们不希望将任何可删除的记录发送到 这。但是，由于这些记录实际上并不是坏记录，因此我们希望 过滤掉它们而不是跳过它们。因此，将只收到 可插入和可更新的记录。​​ItemWriter​​​​ItemWriter​​

要筛选记录，可以从 .框架检测 结果是并避免将该项目添加到传递到的记录列表中 这。从结果中引发的异常 跳。​​null​​​​ItemProcessor​​​​null​​​​ItemWriter​​​​ItemProcessor​​

验证输入

ItemReader 和 ItemWriters 一章讨论了分析输入的多种方法。 如果每个主要实现不是“格式正确”，则会引发异常。如果缺少一系列数据，则会引发异常。同样地 尝试访问 OR 中不存在的索引，或 的格式与预期的格式不同，会导致引发异常。全部 这些类型的异常在返回之前引发。但是，它们没有解决 返回的项目是否有效的问题。例如，如果其中一个字段 是一个年龄，它不可能是消极的。它可以正确解析，因为它存在并且 是一个数字，但不会导致异常。由于已经有过多的 验证框架，Spring Batch 不会尝试提供另一个。相反，它 提供一个简单的接口，称为 ，您可以通过任意数量的 框架，如以下接口定义所示：​​FixedLengthTokenizer​​​​RowMapper​​​​FieldSetMapper​​​​read​​​​Validator​​

public interface Validator<T> {

    void validate(T value) throws ValidationException;

}

协定是该方法在对象无效时引发异常 如果有效，则正常返回。Spring Batch 提供了一个 ，如以下 Bean 定义所示：​​validate​​​​ValidatingItemProcessor​​

XML 配置

<bean class="org.springframework.batch.item.validator.ValidatingItemProcessor">
    <property name="validator" ref="validator" />
</bean>

<bean id="validator" class="org.springframework.batch.item.validator.SpringValidator">
    <property name="validator">
        <bean class="org.springframework.batch.sample.domain.trade.internal.validator.TradeValidator"/>
    </property>
</bean>

爪哇配置

@Bean
public ValidatingItemProcessor itemProcessor() {
  ValidatingItemProcessor processor = new ValidatingItemProcessor();

  processor.setValidator(validator());

  return processor;
}

@Bean
public SpringValidator validator() {
  SpringValidator validator = new SpringValidator();

  validator.setValidator(new TradeValidator());

  return validator;
}

您还可以使用 来验证批注的项目 Bean Validation API （JSR-303） 注释。例如，考虑以下类型：​​BeanValidatingItemProcessor​​​​Person​​

class Person {

    @NotEmpty
    private String name;

    public Person(String name) {
     this.name = name;
    }

    public String getName() {
     return name;
    }

    public void setName(String name) {
     this.name = name;
    }

}

您可以通过在 应用程序上下文，并在面向块的步骤中将其注册为处理器：​​BeanValidatingItemProcessor​​

@Bean
public BeanValidatingItemProcessor<Person> beanValidatingItemProcessor() throws Exception {
    BeanValidatingItemProcessor<Person> beanValidatingItemProcessor = new BeanValidatingItemProcessor<>();
    beanValidatingItemProcessor.setFilter(true);

    return beanValidatingItemProcessor;
}

容错

回滚区块时，读取期间缓存的项目可能是 重新处理。如果将步骤配置为容错（通常通过使用跳过或 重试处理），任何使用的方法都应以 幂等。通常，这将包括不对输入项执行任何更改 和仅更新 实例，即结果。​​ItemProcessor​​​​ItemProcessor​​