如果出现异常，如何覆盖 Spring Batch CompositeItemWriter 管理委托编写者的事务？答案

【问题标题】：How to override Spring Batch CompositeItemWriter manage transaction for delegate writers in case of exception arises?如果出现异常，如何覆盖 Spring Batch CompositeItemWriter 管理委托编写者的事务？
【发布时间】：2020-08-07 06:00:39
【问题描述】：

我在这里扩展这个How does Spring Batch CompositeItemWriter manage transaction for delegate writers? 问题：

在我的情况下，我有一个下面的CompositeItemWriter，它将数据写入同一数据库的多个表中，在写入数据之前它通过实现各种业务规则来转换数据。在这里，一条记录可能满足不同的业务规则等。因此，一位作者可能获得比其他人更多的数据。

@Bean
public CompositeItemWriter<Employee> EmployeeCompositeWriter() throws Exception {
    List<ItemWriter<? super Employee>> employee = new ArrayList<>();
    employee.add(employeeWriter());
    employee.add(departmentWriter());
    employee.add(stockWriter());
    employee.add(purchaseWriter());

    CompositeItemWriter<Employee> compositeItemWriter = new CompositeItemWriter<>();
    compositeItemWriter.setDelegates(employee);
    compositeItemWriter.afterPropertiesSet();
    return compositeItemWriter;
}

场景 - 假设第一作者工作得很好，第二作者产生异常，然后第三和第四作者没有被调用这是由于事务滚动而在 Spring Batch 中默认的Automic 性质返回。

这里即使第二个写入器出现任何异常，我想成功调用第三个和第四个写入器并保存数据，我也想成功保存第一个写入器和第二个写入器的数据..只有我想要的异常数据在SkipListener 的帮助下存储到错误表中，以识别哪些记录是垃圾或垃圾。

解决方案 - 为了实现上述场景，我们在每个写入器的写入方法上添加了@Transactional(propagation = Propagation.REQUIRES_NEW)，第一个写入器现在保存数据，第二个写入器生成异常（使用namedJdbcTemplate.batchUpdate() 批量更新数据) 我们正在缓存它并重新抛出它，但我们可以看到 提交级别降低到 1（偏离路线以识别准确的垃圾记录）并且当第二个作者再次出现异常时，第一个作者正在获取调用它并保存重复的数据，并调用第 2、3 和 4 个写入器，但垃圾记录也不会流向第 3 个和第 4 个写入器。

在这里，如果单个或多个记录是垃圾，我不希望整个批处理作业停止，因为此作业对于我们每次运行都至关重要。如果我们可以保存所有没有出现异常的数据并且仅在SkipListener的帮助下将异常数据保存到错误表中，有什么办法吗？

如果我们可以将批处理组件（例如（READER 或 PROCESSOR）部分）重用到另一个步骤中，有什么办法吗？

【问题讨论】：

什么都没有被跳过，那么为什么skiplistener会在这种情况下提供帮助？似乎您最好编写自己的单一作家，一次完成您想要的事情。从批处理的角度来看，只有一个写入器（不是多个），并且一切都应该（通常）成功与否（这就是写入批处理记录的本质）。
从批次的角度来看，没有任何内容被跳过。跳过是完全跳过的写入，而不是您想要的部分写入。这不是skiplistener的工作方式。

标签： spring spring-batch

【解决方案1】：

我看不出有一种方法可以让 spring-batch 的单个事务将整个块写入原子性与您将原子性保留给单个写入者的想法，只要您愿意skiplistener。

我不确定这是否可行，但也许您可以快速测试它。这就是消息在某些集成框架（如骆驼）中将异常从一个处理器传递到错误处理流程的方式。

您的项目阅读器应该返回一个EmployeeWrapper，其中包含employee 记录，并有一个存储异常的字段。
您的 CompositeItemWriter 收到 List<EmployeeWrapper> 并且复合编写器有 5 个编写器而不是 4 个。第 5 个编写器将执行您的 SkipListener 会执行的操作。

    List<ItemWriter<? super EmployeeWrapper>> employee = new ArrayList<>();
    employee.add(employeeWriter());
    employee.add(departmentWriter());
    employee.add(stockWriter());
    employee.add(purchaseWriter());
    employee.add(errorRecordWriter());

您的前 4 个独立编写者从不抛出异常，而是将其标记为已处理，但将捕获的异常添加为 EmployeeWrapper 的属性。
您的第 5 个errorRecordWriter 接收所有记录，检查任何添加了异常属性的记录并将它们写入错误表。如果写入错误记录失败，可以抛出异常，所有5个写入器都将被重试。
关于批量更新失败时如何知道哪条记录是错误记录。似乎当块中发生错误时，弹簧回滚该块并开始在该块中逐条记录重试它，以便它知道哪条记录是有问题的。所以你可以在你的个人作家身上做同样的事情。即捕获批量更新异常，然后一一重试，将错误记录分开

【讨论】：

我正在使用自定义写入器批量更新 2000 块中的记录。 JdbcTemplate/NamedParameter JdbcTemplate 不会告诉哪些是垃圾/垃圾数据，除非我不扔它。我怎样才能在这里使用错误记录？
嗨 Kavitha - 在我们的例子中，我们有两个不同的写入者将数据插入到同一个表中，因为我们导出的数据不同。现在我们已经将这 2 个写入器组合在一起，它导致了 Spring Batch 提供的平滑事务工作 OOTB 假设一个写入器的 STOCK_ID = 1 失败，那么该项目对于所有其他写入器都失败，因为我们使用 SkipListener 跳过记录，这是我们得到一致的行为。这里我们不需要使用 Custom CompositeItemWriter ，它与内置的 API 配合得很好。感谢您的大力帮助，请恢复！

【解决方案2】：

这里有几件事：

不要将 @Transactional 与 Spring Batch 一起使用 - Spring Batch 会为您管理事务，因此使用该注释会导致问题。不要使用它。
自行管理异常 - 在您描述的场景中，您想为同一个项目调用四个 ItemWriter 实现，但希望在委派的 ItemWriter 级别跳过异常，您将需要编写自己的CompositeItemWriter 实现。 Spring Batch 提供了这种级别的组合（我们将相同的项目委托给每个ItemWriter 实现），但从框架的角度来看，它只是一个ItemWriter。为了在子 ItemWriter 级别处理异常，您需要编写自己的包装器并自己管理异常。

更新：
我指的是自定义ItemWriter 的示例实现（注意下面的代码未经测试）：

public class MyCompositeItemWriter<T> implements ItemWriter<T> {
      private List<ItemWriter<? super T>> delegates;
 
    @Override
      public void write(List<? extends T> items) throws Exception {
            for(ItemWriter delegate : delegates) {
               try {
                  delegate.write(items);
               }
               catch (Exception e) {
                  // Do logging/error handling here
               }
            }
    }

    @Override
    public void setDelegates(List<ItemWriter<? super T>> delegates) {
        super.setDelegates(delegates);
        this.delegates = delegates;
    }
}

【讨论】：

如果我不输入@Transactional(propagation=REQUIRED_NEW)，那么如果第一作者未能保存记录，第二作者将永远不会被调用。所以我认为这是必要的。是的，我开发了 Custom CompositeItemWriter，我需要在其中查找提交间隔以决定成功方案并管理委托的索引位置，以便让 Custom Composite ItemWriter 在早期委托成功时从该委托开始循环。跨度>
再一次，它不是。您需要在您的委托中捕获异常。如果您使用该注释，它将破坏事务与 Spring Batch 的工作方式。
我已经添加了我所有的发现github.com/mminella/scaling-demos/issues/6，请您验证一下吗？请做必要的事情。
BatchUpdate 没有提供哪些记录是垃圾的工具，因此我得到了所有 500 条记录，这是我的块大小。我没有办法识别哪些是垃圾记录，即使我不抛出异常，也不会调用其他作者。您能否展示一些代码，我们可以在这里做得更好吗？我同意玩交易是非常复杂和冒险的，但我需要代码来更多地了解你的 cmets。如果这看起来不错，也请在这里指导 github.com/mminella/scaling-demos/issues/6
@mminella - 在我们的例子中，我们有两个不同的编写者将数据插入到同一个表中，因为我们导出的数据不同。现在我们已经将这 2 个写入器组合在一起，它导致了 Spring Batch 提供的平滑事务工作 OOTB 假设一个写入器的 STOCK_ID = 1 失败，那么该项目对于所有其他写入器都失败，因为我们使用 SkipListener 跳过记录，这是我们得到一致的行为。这里我们不需要使用 Custom CompositeItemWriter ，它与内置的 API 配合得很好。感谢您的大力帮助，请恢复！

【解决方案3】：

问题的主要原因是，我们试图使用两个不同的ItemWriter 将数据写入同一个表中，这导致 Transaction 行为异常。

我们已经实现了SkipListenets（考虑到我们在初始数据加载时执行验证时可能不会经常获取垃圾或垃圾数据。）

由于我们在批处理作业中实施了“Spring Batch Skip Technique”，这有助于我们指定某些异常类型和最大数量。跳过的项目，并且每当抛出这些可跳过的异常之一时，批处理作业不会失败，而是跳过该特定项目并进入下一个项目。只有当最大没有。达到跳过的项目，批处理作业将失败。我们使用了 Spring Batch 的 “容错” 特性的跳过逻辑，应用于面向块的步骤中的项目，而不是整个步骤。

因此，如果 Item 未能在一个委托上写入，那么它将被认为对所有其他委托都失败（该项目不会传递给另一个委托），我们对此很好，因为我们正在捕获错误中的详细信息日志表，我们可以在需要时从那里重新处理它。

【讨论】：