使用 RxJava 的状态机？答案

【问题标题】：State machine using RxJava?使用 RxJava 的状态机？
【发布时间】：2017-09-07 03:24:49
【问题描述】：

我正在尝试全神贯注于 RxJava 并解决我遇到的这个问题，但它似乎非常不适合它，因为 RxJava 似乎不想处理任何类型的状态，而只是传递事件并对它们进行变异以处理它们。

我试图用 RxJava 模拟的基本状态机行为是这样的：

在应用启动事件中，等待下一次应用暂停。
在应用暂停事件时，启动 15 分钟计时器并等待下一个应用恢复。
- 如果应用程序在计时器内恢复，请将其取消并返回到第 1 步。
在应用恢复事件中，如果 15 分钟计时器已过，请刷新并返回第 1 步。

【问题讨论】：

我们实现一些类似的东西的方式是使用RX作为一种单向事件总线，应用程序将事件放在总线上，不同的观察者关心不同的事件，每个观察者保持自己的自己的状态。这就是我们的跟踪架构，有超过一百个不同的事件和几十个观察者，每个观察者处理不同的跟踪事件——有时针对不同的跟踪平台。我认为，如果可能的话，将这个状态保持在 RX 中会过于复杂——尤其是当你考虑到当你的应用程序的内存被系统在 BG 中擦除时，你可能必须保持这个状态。
是的，我认为我们的应用很相似。我们只是在最近才切换，对于 Rx 可能/应该做什么还有很多探索，反之亦然。
我想我们经历了类似的过程，并意识到“RX all the things”并不是一个好方法。除了用于跟踪的真正基本的自制事件总线之外，对我们而言，RX 链只在一个方向上运行——从 Repository 层向上通过 Interactors 到 Presenters，后者直接订阅和调用 View 方法。我们还没有弄清楚新的 ViewModel 和 Lifecycle 架构是否/如何适应它。
只是想注意这里有一个简单的状态机实现：github.com/davidmoten/rxjava2-extras

标签： android rx-java rx-java2

【解决方案1】：

虽然可以在 RxJava 中实现状态机，但它变得非常丑陋，非常快。可以使用switchMap()运算符选择部分分支路径，也可以使用其他几个运算符进行分支合并。

大部分问题在于状态机转换看起来更像go to 而不是结构化编程，而且它们往往根本不像函数式编程。

出于类似的原因，创建任何非平凡状态机的流畅或功能描述是相当困难的。有一个通用状态机的RxJava实现RxFsm，看起来相当称职。您将提供计时器作为外部输入，以及其他类似的变通方法。

这是实现您的状态机的代码，使用了一些 RxJava 元素，并做了一些简化假设：

Observable<AppState> appStateObservable;
AtomicReference<Subscription> timerSubscription = new AtomicReference<>(Subscriptions.empty());

appStateObservable
  .doOnNext( state -> if ( state == START ) doStartup() )
  .filter( state -> state != START )
  .subscribe( state -> if ( state == PAUSE ) {
      timerSubscription.set( Observable.timer(15, MINUTES)
        .subscribe( timerSubscription.get().unsubscribe() ) );
    } else  if ( timerSubscription.get().isUnsubscribed() ) {
      refresh();
    } else {
      timerSubscription.get().unsubscribe();
    } );

它使用timerSubscription 订阅状态来确定计时器是否已触发，从而执行刷新。

【讨论】：

你能告诉我为什么以及如何变得丑陋吗？我在我的项目中实现状态机，但仅在模块/功能方面。我只是想明白你说的丑是什么意思。
“丑陋”来自几个方面：1) FSM 输入来自多个来源并且具有各种类型，2) 状态转换需要移动到观察者链的不同部分，3) 错误处理使上述所有事情变得复杂。