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

Question

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

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

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

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;

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

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

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

Answer 1

即使您现在不需要非阻塞异步调用，Reactor 仍然非常适合这种情况，因为它擅长编排这种处理管道。我认为 Java 8Stream也可以满足要求，但在这方面功能稍弱一些。

为了清楚起见，扩展方法引用，并且通过我的一些猜测，您的代码在 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 代码共享的线程（而不仅仅是特定的处理管道） ）并且资源有限。