java - 在提供给 flatMap 的 RxJava Func1 对象中使用可变状态是个好主意吗？

Question

假设我们需要以某种方式转换一个热 Observable，我们需要知道它之前发出的所有项目才能确定接下来要发出什么。我发现最方便的解决方案是将 Func1 子类的实例传递给 flatMap，它具有全局状态(例如，以前发出的项目的 map 或列表)。在每次调用中，Func1 实例都会更新其状态，并根据该状态决定返回什么。

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

其他可能的解决方案可能是:

1. 创建运算符(operator)。我猜 Operators 中的可变状态是允许的。无论如何，我尽量避免自定义运算符，因为它们更棘手。

2. 通过扫描(在列表或 map 中)传播 Observable 的历史记录。我要么对每个发出的项目使用相同的对象(List 或 Map)，这会在流中引入一个可变对象，要么每次都复制整个对象，这会浪费很多性能。

3. 订阅原始 Observable，修改订阅者的一些全局状态，并使用此全局状态在 Subject(转换后的 Observable)上发射项目。我想到这个是因为它在处理全局状态(和同步)时似乎退出了 RxJava 的范围。

所以问题是:我是否应该在 flatMap 中使用具有可变状态的 Func1 实现来根据先前发出的项目的历史记录来转换项目(顺便说一句，这是有效的)，如果不是，我应该使用什么替代方案？总的来说，我对处理 Observable 转换所需的复杂可变状态的推荐方法感到困惑。

我希望我已经清楚地表达了我的问题。否则，请告诉我，我会尝试借助一些具体问题和代码来描述它。

Answer 1

通常不推荐包含可变状态的函数流，因为可变状态可能会在多个 Subscriber 之间共享到特定的 Observable 链。不过，大多数开发人员通常会在需要时组装 Observable 并且很少重用相同的 Observable。例如，一个按钮点击处理程序将创建一个 Observable，通过组合， fork 出另外两个 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 字段。