multithreading - rxjava2 是否在线程安全方面受到过度保护？

Question

我已经使用 RxJava2 很长时间了。最近在研究RxJava2的源码，试图理解它，写自己的算子实现。我发现有很多线程安全代码。但是我觉得有些是多余的，增加了学习的难度。

根据Reactive Streams 规范2.7 “订阅者必须确保对其订阅的所有调用都来自同一线程或提供相应的外部同步。”。确保其调用是线程安全的不是调用者的职责吗？

但我发现“io.reactivex.internal.operators.flowable.FlowableLimit$LimitSubscriber.request”这个方法使用 AtomicLong 来实现原子性。许多其他代码也是这样做的。为什么？

标签： multithreadingthread-safetyrx-java2

Answer 1

但是我觉得有些是多余的，增加了学习的难度。

在实施运营商时，必须考虑来自不同来源/消费者的各种信号类型的许多可能的并发交互。我花了很多时间研究和优化 RxJava 2。并发性很难，我很确定实际上没有“多余的”实现细节。有些算法看起来像它们的样子总是有原因的。然而，可能有一些优化的可能性，但必须仔细分析和考虑。

根据 Reactive Streams 规范 2.7

考虑到RxJava和 Project Reactor 等实际实现的结果，已经并且目前正在对规则的措辞进行一些讨论和改进。

确保其调用是线程安全的不是调用者的职责吗？

在调用方法Publisher时，特定的运算符通常处于确保线程安全的最佳位置。Subscriber.onXXX否则，aSubscriber将需要更多地了解其来源是什么或做什么。

但我发现“io.reactivex.internal.operators.flowable.FlowableLimit$LimitSubscriber.request”

有很多方法可以确保线程安全。措辞“同步”并不一定意味着synchronized必须使用 Java 关键字。RxJava 主要使用所谓的无锁算法，利用java.util.concurrent.atomic类提供的原子操作。请求管理通常是无锁原子的“客户端”。