【发布时间】：2022-01-08 12:30:05
【问题描述】：
我正在使用 reactor 框架编写一个简单的编排框架，该框架按顺序执行任务，下一个要执行的任务取决于先前任务的结果。根据先前任务的结果，我可能有多种路径可供选择。早些时候，我基于静态 DAG 编写了一个类似的框架，其中我将任务列表作为可迭代对象传递并使用 Flux.fromIterable(taskList)。但是，由于静态数组发布者，这并没有给我动态的灵活性。
我正在寻找像 do(){}while(condition) 这样的替代方法来解决 DAG 遍历和任务决策，我想出了 Flux.generate()。我评估生成方法的下一步并将下一个任务传递到下游。我现在面临的问题是，Flux.generate 不会等待下游完成，而是推送直到条件设置为无效。到任务 1 执行时，任务 2 将被推送n 次，这不是预期的行为。
有人可以指点我正确的方向吗？
谢谢。
使用任务列表（静态 DAG）的第一次迭代
Flux.fromIterable(taskList)
.publishOn(this.factory.getSharedSchedulerPool())
.concatMap(
reactiveTask -> {
log.info("Running task =>{}", reactiveTask.getTaskName());
return reactiveTask
.run(ctx);
})
// Evaluates status from previous task and terminates stream or continues.
.takeWhile(context -> evaluateStatus(context))
.onErrorResume(throwable -> buildResponse(ctx, throwable))
.doOnCancel(() -> log.info("Task cancelled"))
.doOnComplete(() -> log.info("Completed flow"))
.subscribe();
尝试动态dag
Flux.generate(
(SynchronousSink<ReactiveTask<OrchestrationContext>> synchronousSink) -> {
ReactiveTask<OrchestrationContext> task = null;
if (ctx.getLastExecutedStep() == null) {
// first task;
task = getFirstTaskFromDAG();
} else {
task = deriveNextStep(ctx.getLastExecutedStep(), ctx.getDecisionData());
}
if (task.getName.equals("END")) {
synchronousSink.complete();
}
synchronousSink.next(task);
})
.publishOn(this.factory.getSharedSchedulerPool())
.doOnNext(orchestrationContextReactiveTask -> log.info("On next => {}",
orchestrationContextReactiveTask.getTaskName()))
.concatMap(
reactiveTask -> {
log.info("Running task =>{}", reactiveTask.getTaskName());
return reactiveTask
.run(ctx);
})
.onErrorResume(throwable -> buildResponse(ctx, throwable))
.takeUntil(context -> evaluateStatus(context, tasks))
.doOnCancel(() -> log.info("Task cancelled"))
.doOnComplete(() -> log.info("Completed flow")).subscribe();
上述方法的问题是，在执行任务 1 时，onNext() 订阅者打印了很多次，因为generate 正在发布。我希望生成方法等待上一个任务的结果并提交新任务。在非响应式世界中，这可以通过简单的 while() 循环来实现。
每个任务都会执行以下操作。
public class ResponseTask extends AbstractBaseTask {
private TaskDefinition taskDefinition;
final String taskName;
public ResponseTask(
StateManager stateManager,
ThreadFactory factory,
) {
this.taskDefinition = taskDefinition;
this.taskName = taskName;
}
public Mono<String> transform(OrchestrationContext context) {
Any masterPayload = Any.wrap(context.getIngestionPayload());
return Mono.fromCallable(() -> stateManager.doTransformation(context, masterPayload);
}
public Mono<OrchestrationContext> execute(OrchestrationContext context, String payload) {
log.info("Executing sleep for task=>{}", context.getLastExecutedStep());
return Mono.delay(Duration.ofSeconds(1), factory.getSharedSchedulerPool())
.then(Mono.just(context));
}
public Mono<OrchestrationContext> run(OrchestrationContext context) {
log.info("Executing task:{}. Last executed:{}", taskName, context.getLastExecutedStep());
return transform(context)
.doOnNext((result) -> log.info("Transformation complete for task=?{}", taskName);)
.flatMap(payload -> {
return execute(context, payload);
}).onErrorResume(throwable -> {
context.setStatus(FAILED);
return Mono.just(context);
});
}
}
编辑 - 来自 @Ikatiforis 的推荐 - 我得到以下输出
Here's the output from my side.
2021-12-02 09:58:14,643 INFO (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$5:98] On next => Task1
2021-12-02 09:58:14,644 INFO (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$6:101] Running task =>Task1
2021-12-02 09:58:14,644 INFO (reactive_shared_pool) [AbstractBaseTask run:75] Executing task:Task1. Last executed:Task1
2021-12-02 09:58:14,658 INFO (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$5:98] On next => Task2
2021-12-02 09:58:14,659 INFO (reactive_shared_pool) [AbstractBaseTask lambda$run$0:83] Transformation complete for task=?Task1
2021-12-02 09:58:14,659 INFO (reactive_shared_pool) [ResponseTask execute:41] Executing sleep for task=>Task1
2021-12-02 09:58:15,661 INFO (reactive_shared_pool) [AbstractBaseTask lambda$run$4:106] Success for task=>Task1
2021-12-02 09:58:15,663 INFO (reactive_shared_pool)
[ReactiveEngine lambda$doOrchestration$6:101] Running task =>Task2
2021-12-02 09:58:15,811 INFO (cassandra-nio-worker-8) [AbstractBaseTask run:75] Executing task:Task2. Last executed:Task2
2021-12-02 09:58:15,811 INFO (reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$5:98] On next => Task2
2021-12-02 09:58:15,812 INFO (reactive_shared_pool) [AbstractBaseTask lambda$run$0:83] Transformation complete for task=?Task2
2021-12-02 09:58:15,812 INFO (reactive_shared_pool) [ResponseTask execute:41] Executing sleep for task=>Task2
2021-12-02 09:58:15,837 INFO (centaurus_reactive_shared_pool) [ReactiveEngine lambda$doOrchestration$9:113] Completed flow
我在这里看到了几个问题--
The sequence of execution is
1. Task does transformations ( runs on Mono.fromCallable)
2. Task induces a delay - Mono.fromDelay()
3. Task completes execution. After this, generate method should evaluate the context and pass on the next task to be executed.
What I see from the output is:
1. Task 1 starts the transformations - Runs on Mono.fromCallable.
2. Task 2 doOnNext is reported - which means the stream already got this task.
3. Task 1 completes.
4. Task 2 starts and executes delay -> the stream does not wait for response from task 2 but completes the flow.
【问题讨论】：
-
上下文中的状态是否在任务完成之前更新？这可以解释为什么最后一项任务似乎被缩短了（也许是因为
takeUntil？）。如果您需要处理所有任务及其结果，为什么要使用takeUntil？看起来经典的完成就足够了...... -
@SimonBaslé 状态会根据任务错误或逻辑故障进行更新。所以它应该等待所述任务的完成，因为它是链式的。 takeUntil 接受一个方法，该方法基本上查看这个状态标志，一个二进制标志，并终止流程。如果逻辑输出为假，我们不需要继续。所以经典补全不是一种选择。
-
@SimonBaslé 同样在这种情况下，我将代码修改为 Ikatiforis 的建议 - 预取使流执行相同的任务两次，而我的预期行为是 - 生成一个任务 -> 等待直到它完成 -> 评估状态和下一步 -> 传递任务。 （这是 Flux.generate 的功能）
标签： java reactive-programming spring-webflux project-reactor