RxJava 共有三个大版本：

1. Version 1.x

   The 1.x version is end-of-life as of March 31, 2018. No further development, support, maintenance, PRs and updates will happen. The Javadoc of the very last version, 1.3.8, will remain accessible.

2. Version 2.x

   The 2.x version is end-of-life as of February 28, 2021. No further development, support, maintenance, PRs and updates will happen. The Javadoc of the very last version, 2.2.21, will remain accessible.

3. Version 3.x (Javadoc)

   • Single dependency: Reactive-Streams.
   • Java 8+ or Android API 21+ required.
   • Java 8 lambda-friendly API.
   • Android desugar friendly.
   • Fixed API mistakes and many limits of RxJava 2.
   • Intended to be a replacement for RxJava 2 with relatively few binary incompatible changes.
   • Non-opinionated about the source of concurrency (threads, pools, event loops, fibers, actors, etc.).
   • Async or synchronous execution.
   • Virtual time and schedulers for parameterized concurrency.
   • Test and diagnostic support via test schedulers, test consumers and plugin hooks.
   • Interop with newer JDK versions via 3rd party libraries, such as
   • Java 9 Flow API
   • Java 21 Virtual Threads
   • Learn more about RxJava in general on the Wiki Home.

1. RxJava 基础

• io.reactivex.rxjava3.core.Flowable: 0..N flows, supporting Reactive-Streams and backpressure
• io.reactivex.rxjava3.core.Observable: 0..N flows, no backpressure,
• io.reactivex.rxjava3.core.Single: a flow of exactly 1 item or an error,
• io.reactivex.rxjava3.core.Completable: a flow without items but only a completion or error signal,
• io.reactivex.rxjava3.core.Maybe: a flow with no items, exactly one item or an error.

1.1 Observable

Observable 是 Rx 中的核心抽象，代表一个可观察的对象。其产生数据流，并可以结合各种操作符来改变数据流。

常用的创建 Observable 方法有两种：

1. 使用 fromXxx(), just() 等方法从现有对象中创建;
2. 使用 Observable#create() 来自定义数据流；

1
2
3
4
5
6
7
8

Observable<Integer> just = Observable.just(10);
Observable<Integer> integerObservable = Observable.fromArray(1, 2, 3, 4, 5);
Observable<String> stringObservable = Observable.<String>create(c -> {
    c.onNext("hello");
    c.onNext("world");
    c.onNext("this");
    c.onComplete();
});

create() 方法的参数为 ObservableOnSubscribe 类型，是一个以 Emitter 为参数的函数。当有订阅者订阅 Observable 时，ObservableOnSubscribe 表示的方法便会执行一次，以产生数据。

考虑到在 ObservableOnSubscribe 中，如上 c.onNext("hello"), 便是直接调用订阅者的回调事件，这是同步调用的。所以得出结论：观察者的监听方法调用是由数据发送的那个线程执行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// ObservableCreate.java 源码, onNext 是同步执行的
static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable {
    CreateEmitter(Observer<? super T> observer) {
        this.observer = observer;
    }

    // Emitter 中的 onNext 数据发送
    @Override
    public void onNext(T t) {
        if (t == null) {
            onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));
            return;
        }
        if (!isDisposed()) {
            // 同步调用 Observer 的 onNext 监听方法
            observer.onNext(t);
        }
    }
}

1.2 Observer

Observer 观察者，通过实现三个方法来响应 Observable 数据流

• onNext(data) 处理数据
• onError(error) 处理异常，然后中断流
• onComplete() 标记流的完成

2. Observable 的几个特点

每次订阅时，create 方法都会执行一遍，特殊情况下会有问题，如使用 http 访问远程数据时，每个观察者都会创建自己独立的 http 客户端，这会导致资源浪费。

Subject 可以解决这个问题，Subject 同时实现了 Observable 和 Observer 接口，即是观察者，又是一个可观察对象。Subject 会维护一个观察者列表。 通过引用 Subject，然后订阅真正的数据源，把自身当作一个中介，把上游的数据源发布到下游。

介绍几种 Subject

• PublishSubject，实时的把上游收到的数据，转发到全部的订阅者。

• BehaviorSubject, 保留最近的一个数据给新观察者，后接收的数据实时转发到订阅者

• ReplaySubject, 重复所有历史数据给新观察者,

• AsyncSubject, 当上游数据完成时，取最后一个转发给订阅者

3. 操作符

Operators List

Creating

Transforming

Filtering

Combining

Error Handling

Utility

Conditional and Boolean

Mathematical and Aggregate

Connectable Observable

4. 背压 Backpressure

rxjava 中，可以通过使用 buffer, window 等操作符来缓存部分数据，已达到批量处理操作，抵御数据洪峰。但真正灵活的还是背压功能，因为只有消费者自己知道自己的消费速度。

背压，意思为当消费者的消费速度跟不上生产者的生产速度时，消费端提供一种反馈通道来通知生产者调整发射速度

Flowable 类似于 Observalbe，但支持 backpressure 功能

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

Flowable.range(0, 1_000_000_000)
    .subscribe(object: DefaultSubscriber<Int>() {
        override fun onStart() {
            request(10)
        }
        override fun onNext(p0: Int) {
            println(p0)
            TimeUnit.SECONDS.sleep(1)
            request(10)
        }

        override fun onError(p0: Throwable?) {
            println(p0?.message)
        }

        override fun onComplete() {
            println("Complete")
        }
    })

通过实现 DefaultSubscriber 的 onStart 方法，在 onStart 中使用 request 初始化消费数量(默认 Long.MAX_VLAUE)，然后在每次 onNext 中在使用 request 请求获取消息（这样把本是 push 模型的 rxjava 变成了 pull 模型）

rxjava 中 Flowable 创建的源都支持背压操作。但如果想自定义源，需要自己实现背压支持，需要在 Flowable 的 Emitter 支持 requested() 方法，可以知道下游所设定的 request 值。

下列我们创建一个可持续发送数据的源，并使用 requested 和 request 方法来自定义背压操作

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

    fun customBackpressureSubscribe() {
        Flowable.create<String>({
            for(i in 1..1000) {
                while (it.requested().toInt() == 0) {
                    if(it.isCancelled) {
                        return@create
                    }
                }
                println("Subscriber request ${it.requested()}")
                it.onNext(i.toString())
            }
            it.onComplete()
        }, BackpressureStrategy.ERROR)
            .observeOn(Schedulers.computation())
            .subscribe(object: DefaultSubscriber<String>() {
                override fun onStart() {
                    request(1)
                }
                override fun onNext(p0: String) {
                    println(p0)
                    request(1)
                    TimeUnit.MILLISECONDS.sleep(500)
                }

                override fun onError(p0: Throwable?) {
                    TODO("Not yet implemented")
                }

                override fun onComplete() {
                    TODO("Not yet implemented")
                }

            })
    }