问题描述
我正致力于将阻塞式顺序编排框架转换为响应式。目前,这些任务是动态的,并通过 JSON 输入输入引擎。引擎拉取类并执行 run()
方法,并将状态与每个任务的响应一起保存。
如何在反应堆中实现相同的链接?如果这是一个静态 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,并且应该停止通量。
解决方法
当然,我们可以:
- 从任务列表中创建通量(如果适合被动地生成任务列表,那么您可以直接用通量替换该数组列表,如果没有,则保持原样);
-
flatMap()
每个任务到您的task.run()
方法(根据问题现在返回一个Mono
; - 确保我们只在
evaluateResult()
为真时消耗元素; - ...然后最后像以前一样返回
SUCCESS
状态。
因此,将所有这些放在一起,只需将循环和返回语句替换为:
Flux.fromIterable(tasks)
.flatMap(task -> task.run(ctx))
.takeWhile(stepContext -> evaluateResult(stepContext))
.then(Mono.just(Status.SUCCESS));
(由于我们已将其设为响应式,因此您的方法显然需要返回 Mono<Status>
而不仅仅是 Status
。)
根据评论更新 - 如果您只想“一次执行一个”而不是同时执行多个,您可以使用 concatMap()
而不是 flatMap()
。