Java 应用程序：序列工作流模式

Question

我有一个 Spring Web 应用程序。当用户调用保存端点时，系统应该执行许多外部调用以将状态保存在多个微服务中。但是，这些步骤相互依赖。换句话说，我有一系列步骤要执行。 sequence pattern

只是一个一个地调用一组步骤没什么大不了的，我可以为每个步骤创建一个类，然后一个一个地调用它们，在步骤之间进行适当的修改。

但是，每个步骤都可能失败，如果发生这种情况，应正确报告给用户。这是一个直接解决方案的伪代码：

var response = new Response()
try {
    var result1 = step1.execute(args1)
    var args2 = process(result1, args1)
    var result2 = step2.execute(args2)
    ...
catch(Step1Exception e) {
    response.setIsPartialSuccess(true);
    response.setPartialResults(e.getDetails())
}
catch(Step2Exception e) {
    response.setIsPartialSuccess(true);
    response.setPartialResults(e.getDetails())
}
return response; 

每个步骤都可以处理项目列表。有些步骤会一次发送所有项目（要么全部失败，要么都不发送），有些步骤会一个一个地发送它们（一半可以失败，一半可以通过）。 StepException 将包含该信息，即通过什么，什么失败。

如您所见，它并不是真正可维护的。在这里使用 Spring Batch 会有点矫枉过正，因为我不是在读写东西，我不需要任何多线程、作业细节或检查点。不过思路很相似，我想创建一些积木并控制流程。

目前我正试图弄清楚 Spring Reactor 是否可以在这里提供帮助（是的，我知道它用于不同的目的），因为它具有带有一些错误处理的流/管道。想象一下我可以写这样的东西：

var context = new Context(response, args1);
Mono.just(context)
    .map(step1::execute)
    .onErrorReturn(e -> context.withError(e))
    //I assume if error happened before
    //steps below are not executed
    .map(step2::execute) 
    .onErrorReturn(e -> context.withError(e))
    .block()
 return context;

您可以将反应式应用程序处理的数据视为通过装配线移动。 Reactor 既是传送带，又是工作站。原材料从来源（原始发布者）倾泻而出，最终成为准备好推送给消费者（或订阅者）的成品。

原材料可以经过各种转换和其他中间步骤，或者成为将中间部件聚合在一起的更大装配线的一部分。如果在某一点出现故障或堵塞（也许装箱产品需要很长时间），受影响的工作站可以向上游发出信号以限制原材料的流动。

换句话说，我正在寻找与上述类似的框架。我现在不需要任何异步处理或重试，但它们将来可能有用。请告诉我是否有比反应堆更好的东西来满足我的需求。

Answer 1

即使您现在不需要非阻塞异步调用，Reactor 仍然可以很好地解决这个问题，因为它擅长编排这种处理管道。我认为 Java 8 Stream 也可以满足要求，但在这方面功能稍差。

为了清楚起见，扩展了方法引用，加上我的一些猜测，您的代码在 Reactor 中看起来像这样：

var response = Mono.just(initialArgs)
    .flatMap(args1 -> Mono.fromCallable(() -> step1.execute(args1))
        .map(result1 -> process(result1, args1) //args1 still in scope inside flatMap
    )
    .flatMap(args2 -> Mono.fromCallable(() -> step2.execute(args2))
    //alternatively to last flatMap, with caveat:
    //.map(args2 -> step2.execute(args2))
    .map(endResult -> new Response(endResult))
    .onErrorResume(error -> {
        Response errorResponse = new Response();
        errorResponse.setIsPartialSuccess(true);
        errorResponse.setPartialResults(error.getDetails());
        return Mono.just(errorResponse);
    })
    .block();

在此特定链中使用的运算符不会更改线程，因此这将全部在调用最后一个 block() 方法的线程中执行。

任何步骤的错误都会停止整个处理并传播到最后（block() 会抛出异常）。

请注意，某些运算符（主要是那些有时间概念的运算符）会更改线程，此时 stepX.execute 被阻塞成为一个问题，因为这会阻塞应该由整个 Reactor 代码共享的线程（不仅是一个特定的处理管道）而且是有限的资源。