在给定给FlatMap的RxJava Func1对象中使用可变状态是一个好主意吗？

Question

假设我们需要转换一个hot Observable,我们需要知道它以前发出的所有项,以便能够确定下一步要发出什么。我发现最方便的解决方案是将Func1子类的一个实例传递给FlatMap,该子类具有全局状态（例如,一个映射或先前发出的项列表）。在每个调用中，Func1实例将更新其状态，并根据该状态决定返回什么。

但是，我担心这个解决方案的“美好”。据我所知，RxJava在全局和可变状态方面并不是很好，而这个解决方案似乎与之形成了对比。另一方面，我确信我的Observable实现了Observable契约，因此它似乎至少是一个工作解决方案，如果可以并发调用它，同步将解决问题。

其他可能的解决办法是：

通过扫描（在列表或地图中）传播可观察的历史。我要么对每个发出的项使用相同的对象（列表或映射），这将向流中引入一个可变对象，要么每次都复制整个对象，这将浪费大量性能。

订阅原始的Observable，从订阅服务器修改一些全局状态，并使用该全局状态在主题（转换的Observable）上发出项。我之所以这么想，是因为当它处理全局状态（和同步）时，它似乎退出了RxJava的范围。

所以问题是：我是否应该使用flatMap中具有可变状态的Func1实现来基于先前发出的项的历史来转换项（顺便说一句，它起作用了），如果不是，我应该使用什么替代方案？一般而言，我对处理可观察对象转换所需的复杂可变状态的推荐方法感到困惑。

我希望我已经清楚地表达了我的问题。否则，让我知道，我会尝试在一些具体问题和代码的帮助下描述它。

Answer 1

通常不建议使用包含可变状态的函数的流，因为可变状态可能会在多个订阅者之间共享到特定的可观察链。然而，大多数开发人员通常会在需要时汇编observable，而很少重用相同的observable。例如，按钮单击处理程序将创建一个observable，该observable通过组合将其他两个observable分开，以异步地从两个不同的地方获取数据，然后订阅这个线程本地的observable实例。单击一个新的按钮将使用一个新的独立的observable来重复这个过程。

以下是有状态函数问题的解决方案：使有状态位的存在取决于订阅的单个订阅者:defer()

Observable<Integer> o = Observable.defer(() -> {
    return Observable.range(1, 10)
            .map(new Func1<Integer, Integer>() {
                int sum;
                @Override
                public Integer call(Integer v) {
                    sum += v;
                    return sum;
                }
            });
});

o.subscribe(System.out::println);
o.subscribe(System.out::println);

由于将为每个subscribe调用创建func1内部类，因此其sum字段对于每个单独的使用者都是本地的。还请注意，sum将返回并自动装箱到一个不可变的integer，然后可以在其他线程中自由读取它（想想observeon)，因为它完全脱离了sum字段。