RxJava 线程安全答案

【问题标题】：RxJava thread-safetyRxJava 线程安全
【发布时间】：2013-09-29 17:55:17
【问题描述】：

这段代码是线程安全的吗？

Observable<String> observable = ... // some observable that calls
                                    // onNext from a background thread

observable
  .scan(new ArrayList<String>(), (List<String> acc, String next) -> {
    acc.add(next);
    return acc;
  })
  .subscribe( list -> {
    // do somethind with sequence of lists
    ...
  });

我很好奇，因为 ArrayList 不是线程安全的数据结构。

【问题讨论】：

Rx 设计指南很有帮助：go.microsoft.com/fwlink/?LinkID=205219.
Is SerializedSubject necessary for thread-safety in RxJava的可能重复

标签： java multithreading thread-safety rx-java

【解决方案1】：

作为一个快速的答案，在 .NET（原始 Rx 实现）中，可以假设来自可观察序列的所有值都是连续的。这并不排除它是多线程的。但是，如果您以多线程方式生成值，那么您可能希望通过查找与 .NET Synchronize() Rx 运算符等效的函数来强制执行顺序性质。

另一种选择是检查 RxJava 源代码中 Scan 的实现，以验证它确实执行了您希望/期望的顺序特性，从而为您的累加器函数提供安全性。

【讨论】：

RxJava 中的 scan 与 .Net 原始版本完全相同。
干杯。我主要是说，随着时间的推移，我无法确定 .NET 和 Java 代码库是否相同。

【解决方案2】：

如果这段代码不是线程安全的，那么要么 RxJava 被破坏，要么你的 Observable 源被破坏——操作符不可重入是 Rx 合约的一部分。

【讨论】：

如果 RxJava 实际实现了它，这可以通过 Rx 的 Checked() 观察者操作符来强制执行。拉请求，有人吗？
在我看来，块的线程安全并不是那么简单。在帖子中显示的代码中没有包含太多细节。我猜@orionll 没有注意到订阅是默认同步交付的。但是如果转换成异步订阅，那么访问数组列表就不再安全了。
@allprog 你说得对，这并不简单，但这是 Rx 的优势之一，因为它在幕后工作以确保安全。
我的意思是，只要程序员能够在一个可以在全局状态下操作的块中插入任意函数，线程安全就不再是一个可确定的问题，而无需知道完整的语境。 scan 的源代码表明，只要初始列表支持并发或不在全局上下文中访问，它将是安全的。但这个陈述只指那个单一的对象。如果块包含对全局应用程序状态的管理，那么只看块的代码很难确定线程安全。