RxJava 适合分支工作流程吗？

我正在使用 RxJava 来处理我们从队列中提取的一些通知。

RxJava seemed to work fine with a simple workflow, now with new requirements coming in, the flow is growing in complexity with more branches (please see below picture as a reference) I tried to exemplify the flow with a small unit test:

@Test
public void test() {
    Observable.range(1, 100)
        .groupBy(n -> n % 3)
        .toMap(GroupedObservable::getKey)
        .flatMap(m1 -> {
            Observable<Integer> ones1 = m1.get(0);
            Observable<Integer> twos1 = m1.get(1).map(n -> n - 10);
            Observable<Integer> threes = m1.get(2).map(n -> n + 100);
            Observable<Integer> onesAndTwos = Observable.merge(ones1, twos1)
                .map(n -> n * 3)
                .groupBy(n -> n % 2)
                .toMap(GroupedObservable::getKey)
                .flatMap(m2 -> {
                    Observable<Integer> ones2 = m2.get(0).map(n -> n * 10);
                    Observable<Integer> twos2 = m2.get(1).map(n -> n * 100);
                    return Observable.merge(ones2, twos2);
                });
                return Observable.merge(onesAndTwos, threes).map(n -> n +1);
        })
        .subscribe(System.out::println);
}

虽然使用 RxJava 在技术上仍然可以实现，但我现在想知道这是否是一个好的选择，为了形式化分支，我必须在主程序中进行两层嵌套flatMap，这看起来不太整洁。

这是描述上述工作流程的正确方式吗？或者 RxJava 不太适合分支工作流程？

感谢您的帮助！

AFAIK 分组可观察是正确的方法。就个人而言，如果你的图片中“按类型分割”和“合并所有内容”之间的任何内容都是异步的，那么在 RX 中执行此操作肯定有很多优点，例如重试逻辑、缓冲、错误处理、背压等。如果是常规的非异步代码，我想这是个人偏好。您可以使用 RX 来完成此操作，但您也可以使用常规同步代码执行“按类型拆分”和“合并所有内容”之间的所有操作。

无论您选择哪种方式，拆分代码以使其更具可读性始终是一个好主意，这样您就可以像我们读取您附加的图像一样轻松地“读取流程”。