java - 依赖图的响应式(Reactive)执行

Question

我需要创建一个服务(使用 Java)来接受任务图并并行执行它们，同时考虑这些任务之间的依赖关系。

例如，我们有 6 个任务:A、B、C、D、E、F。

依赖关系是:

A -> C

B -> C、D

C -> F

D -> E

E -> F

这将创建(可能的)并行执行组:A+B、C+D、E、F。

如果任务执行不成功(返回错误)，则其依赖的任务将不会被执行。

另一个要求是对任务执行(或失败)产生副作用:通知其他一些服务(这意味着我们也必须将失败任务的依赖项视为失败？)。

处理所有任务后(直接成功或失败+依赖项失败)，我想将此“批处理”标记为已完成(调用另一个服务)。

我考虑采用响应式(Reactive)方法来解决这个问题，并使用 RxJava，因为它具有异步性质。

我对这种方法还很陌生，虽然 zip/switchMap 与 doOnComplete/doOnError 相结合似乎是一个不错的方向，但我不太确定如何在这种情况下使用它们。

很高兴在这里得到一些建议:)

Answer 1

对于根据依赖关系图(严格来说是依赖非循环图)运行 Observables 的查询，您可以使用 concat()和 merge()运算符组成可观察量以按照给定图(DAG)运行。

对于您的示例，以下是如何构建 Observable DAG 来并行执行它们:

package rxtest;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.schedulers.Schedulers;


public class ReactiveDagTest {
    private static final Logger logger = LoggerFactory.getLogger(ReactiveDagTest.class);
    private static Executor customExecutor = Executors.newFixedThreadPool(20);

    @Test
    public void stackOverflowTest() {
        Observable<Character> a = createObservable('A', 100);
        Observable<Character> b = createObservable('B', 150);
        Observable<Character> c = createObservable('C', 500);
        Observable<Character> d = createObservable('D', 200);
        Observable<Character> e = createObservable('E', 300);
        Observable<Character> f = createObservable('F', 400);

        logger.info("BEGIN");

        // As Observable for B is referred at two places in the graph, it needs to be cached to not to execute twice
        Observable<Character> bCached = b.cache();

        Observable.concat(
                Observable.merge(
                        Observable.concat(
                                Observable.merge(a, bCached),
                                c),
                        Observable.concat(bCached, d, e)),
                f)
                .toBlocking()
                .subscribe(i -> logger.info("Executed : " + i));

        logger.info("END");
    }

    private Observable<Character> createObservable(char c, int sleepMs) {
        Observable<Character> single = Observable.just(c)
                .flatMap(i -> Observable.<Character> create(s -> {
                    logger.info("onSubscribe Start Executing : {}", i);
                    sleep(sleepMs);
                    s.onNext(Character.valueOf(i));
                    s.onCompleted();
                }).subscribeOn(Schedulers.from(customExecutor)));
        return single;
    }

    private void sleep(int ms) {
        try {
            Thread.sleep(ms);
        }
        catch (InterruptedException e) {
        }
    }
}

输出将是

22:19:22.107 [main] INFO rxtest.ReactiveDagTest BEGIN
22:19:22.181 [pool-1-thread-1] INFO rxtest.ReactiveDagTest onSubscribe Start Executing : A
22:19:22.181 [pool-1-thread-2] INFO rxtest.ReactiveDagTest onSubscribe Start Executing : B
22:19:22.284 [main] INFO rxtest.ReactiveDagTest Executed : A
22:19:22.333 [main] INFO rxtest.ReactiveDagTest Executed : B
22:19:22.333 [main] INFO rxtest.ReactiveDagTest Executed : B
22:19:22.333 [pool-1-thread-3] INFO rxtest.ReactiveDagTest onSubscribe Start Executing : C
22:19:22.334 [pool-1-thread-4] INFO rxtest.ReactiveDagTest onSubscribe Start Executing : D
22:19:22.534 [main] INFO rxtest.ReactiveDagTest Executed : D
22:19:22.534 [pool-1-thread-5] INFO rxtest.ReactiveDagTest onSubscribe Start Executing : E
22:19:22.833 [main] INFO rxtest.ReactiveDagTest Executed : C
22:19:22.835 [main] INFO rxtest.ReactiveDagTest Executed : E
22:19:22.835 [pool-1-thread-6] INFO rxtest.ReactiveDagTest onSubscribe Start Executing : F
22:19:23.236 [main] INFO rxtest.ReactiveDagTest Executed : F
22:19:23.236 [main] INFO rxtest.ReactiveDagTest END

If a task execution was not successful (returned an error), its dependent tasks will not be executed.

这隐含在上述解决方案中，如果图中的任何节点失败，则图中的其他节点将不会执行。