Java 应用程序 : Sequence workflow pattern

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 就太过分了，因为我不是在读写东西，我不需要任何多线程、作业细节或检查点。但是，想法很相似，我想创建一些构建 block 并控制流程。

目前我正在尝试弄清楚 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;

You can think of data processed by a reactive application as moving through an assembly line. Reactor is both the conveyor belt and the workstations. The raw material pours from a source (the original Publisher) and ends up as a finished product ready to be pushed to the consumer (or Subscriber).

The raw material can go through various transformations and other intermediary steps or be part of a larger assembly line that aggregates intermediate pieces together. If there is a glitch or clogging at one point (perhaps boxing the products takes a disproportionately long time), the afflicted workstation can signal upstream to limit the flow of raw material.

换句话说，我正在寻找一个与上述类似的框架。我现在不需要任何异步处理或重试，但它们将来可能会有用。如果有比 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 代码(不仅是特定的处理管道)组成，而且资源有限。