ReactorJava中反应性任务的顺序执行

Question

我正在将阻塞顺序编排框架转换为反应式。现在，这些任务是动态的，并通过JSON输入输入到引擎中。引擎提取类并执行run（）方法，并使用每个任务的响应保存状态
如何在Reactor中实现相同的链接？如果这是一个静态DAG，我会将它与flatMap或then操作符链接起来，但由于它是动态的，我如何继续执行反应性任务并收集每个任务的输出？

示例：
非反应性接口：

public interface OrchestrationTask {
 OrchestrationContext run(IngestionContext ctx);
}

核心发动机

public Status executeDAG(String id) {
  IngestionContext ctx = ContextBuilder.getCtx(id);
  List<OrchestrationTask> tasks = app.getEligibleTasks(id);

  for(OrchestrationTask task : tasks) {
    // Eligible tasks are executed sequentially and results are collected.
    OrchestrationContext stepContext = task.run(ctx);
    if(!evaluateResult(stepContext)) break;
  }
  return Status.SUCCESS;
}

按照上面的示例，如果我将任务转换为返回Mono

更新：：

反应性任务示例。

public class SampleTask implements OrchestrationTask {  
  @Override
  public Mono<OrchestrationContext> run(OrchestrationContext context) {  
  // Im simulating a delay here. treat this as a long running task (web call) But the next task needs the response from the below call.
  return Mono.just(context).delayElements(Duration.ofSeconds(2));
 }

因此，我将有一系列任务来完成各种事情，但是每个任务的响应依赖于以前的任务，并存储在编排上下文中。每当发生错误时，业务流程上下文标志将设置为false，通量应停止。

Answer 1

当然，我们可以：

• 从任务列表中创建流量（如果以反应方式生成任务列表是合适的，那么可以直接用流量替换该arraylist，如果不是，则保持原样）

所以把所有这些放在一起，替换你的循环

Flux.fromIterable(tasks)
        .flatMap(task -> task.run(ctx))
        .takeWhile(stepContext -> evaluateResult(stepContext))
        .then(Mono.just(Status.SUCCESS));

（因为我们已经使它成为被动的，所以您的方法显然需要返回一个Mono

根据注释进行更新-如果您只是希望“一次执行一个”，而不是同时执行多个，则可以使用concatMap（）而不是flatMap（）。