跳过spring-batch中的异常并在数据库中提交错误

Question

我正在使用 Spring 批处理来编写批处理，但在处理异常时遇到了问题。

我有一个阅读器，它从具有特定状态的数据库中获取项目。阅读器将项目传递给可以启动异常MyException.class 的处理器步骤。当抛出此异常时，我想跳过导致该异常的项目并继续阅读下一个。

这里的问题是我需要更改数据库中该项目的状态，以便读者不会再次获取它。

这是我尝试过的：

return this.stepBuilderFactory.get("name")
            .<Input, Output>chunk(1)
            .reader(reader())
            .processor(processor())
            .faultTolerant()
            .skipPolicy(skipPolicy())
            .writer(writer())
            .build();

在我的 SkipPolicy 类中，我有下一个代码：

public boolean shouldSkip(Throwable throwable, int skipCount) throws SkipLimitExceededException {
    if (throwable instanceof MyException.class) {
        // log the issue
        // update the item that caused the exception in database so the reader doesn't return it again
        return true;
    }
    return false;
}

使用此代码，将跳过异常并再次调用我的阅读器，但是 SkipPolicy 没有提交更改或进行回滚，因此阅读器获取项目并尝试再次处理它。

我也尝试过使用 ExceptionHandler：

return this.stepBuilderFactory.get("name")
            .<Input, Output>chunk(1)
            .reader(reader())
            .processor(processor())
            .faultTolerant()
            .skip(MyException.class)
            .exceptionHandler(myExceptionHandler())
            .writer(writer())
            .build();

在我的 ExceptionHandler 类中，我有下一个代码：

public void handleException(RepeatContext context, Throwable throwable) throws Throwable {
     if (throwable.getCause() instanceof MyException.class) {
        // log the issue
        // update the item that caused the exception in database so the reader doesn't return it again
     } else {
        throw throwable;
     }
}

使用此解决方案，数据库中的状态发生了变化，但它不调用读取器，而是再次调用processor() 的方法process，进入无限循环。

我想我可以在我的步骤中使用侦听器来处理异常，但我不喜欢这种解决方案，因为我必须克隆大量代码，因为我的代码的不同步骤/处理器可能会启动此异常.

我做错了什么？

编辑：经过大量测试并使用不同的侦听器（如 SkipListener）后，我无法实现我想要的，Spring Batch 总是回滚我的 UPDATE。

调试这是我发现的：

一旦调用了我的侦听器并更新了我的项目，程序就会在类FaultTolerantChunkProcessor 中输入方法write（第327 行）。 这个方法会尝试下一段代码（复制自github）：

try {
    doWrite(outputs.getItems());
} catch (Exception e) {
    status = BatchMetrics.STATUS_FAILURE;
    if (rollbackClassifier.classify(e)) {
        throw e;
    }
    /*
     * If the exception is marked as no-rollback, we need to
     * override that, otherwise there's no way to write the
     * rest of the chunk or to honour the skip listener
     * contract.
     */
     throw new ForceRollbackForWriteSkipException(
        "Force rollback on skippable exception so that skipped item can be located.", e);
}

SimpleChunkProcessor 类中的方法 doWrite（第 151 行）将尝试写入输出项列表，但是，在我的情况下，列表为空，因此在第 #159 行（方法writeItems ) 将启动IndexOutOfBoundException，导致ForceRollbackForWriteSkipException 并进行我正在遭受的回滚。

如果我重写 FaultTolerantChunkProcessor 类并且在列表为空的情况下避免写入项目，那么一切都会按预期工作，提交更新并且程序会跳过错误并再次调用读取器。

我不知道这是否真的是一个错误，或者它是由我在代码中做错了什么引起的。

Answer 1

在我看来，SkipListener 比 ExceptionHandler 更适合您的用例，因为它使您可以访问导致异常的项目。使用异常处理程序，您需要在异常或重复上下文中携带项目。

此外，skip listener 可以让您知道异常发生在哪个阶段（即在读取、处理或写入中），而使用异常处理程序，您需要自己找到一种方法来检测它。如果所有阶段的跳过代码都相同，则可以在侦听器的所有方法中调用更新项目状态的相同方法。