雨羊的博客

参考文献：

给 Android 开发者的 RxJava 详解
 RxJava和RxAndroid

仓库：

RxJava
RxAndroid

1. 前言

RxJava 是一个实现异步操作的库，而且很简洁。

2. 基本使用

创建观察者

Observer 即观察者，它决定事件触发的时候将有怎样的行为。 RxJava 中的 Observer 接口的实现方式：

// String 为泛型， 传递的参数的类型
Observer<String> observer = new Observer<String>() {
    public void onSubscribe(@NonNull Disposable disposable) {
        // 订阅的时候回调
    }

    public void onNext(@NonNull String s) {
        // 消费方法
    }

    public void onError(@NonNull Throwable throwable) {
        // 错误发生的时候回调
    }

    public void onComplete() {
        // 完成的时候回调
    }
};

创建被观察者

Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。 RxJava 使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则：

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
        observableEmitter.onNext("Hello");
        observableEmitter.onComplete();
    }
});

可以看到，这里传入了一个 ObservableOnSubscribe 对象作为参数。ObservableOnSubscribe 会被存储在返回的 Observable 对象中，它的作用相当于一个计划表，当 Observable 被订阅的时候，ObservableOnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发（对于上面的代码，就是观察者 observer 将会被调用一次 onNext() 和一次 onCompleted()。这样，由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。

创建了 Observable 和 Observer 之后，再用 subscribe() 方法将它们联结起来，整条链子就可以工作了。代码形式很简单：

// 注意是被观察者去调用 subscribe
observable.subscribe(observer);

3. 进阶使用

create() 方法是 RxJava 最基本的创造事件序列的方法。基于这个方法， RxJava 还提供了一些方法用来快捷创建事件队列，例如：

just(T...)

将传入的参数依次发送出来。

Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

from()

from(T[]) / from(Iterable<? extends T>) : 将传入的数组或 Iterable 拆分成具体对象后，依次发送出来。

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

简单例子

目的：输出 nums 数组的数字

Observable.fromArray(nums).subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println(integer);
    }
});

可以简化为

Observable.fromArray(nums)
        .subscribe(integer -> System.out.println(integer));

4. 线程控制

在 RxJava 的默认规则中，事件的发出和消费都是在同一个线程的。也就是说，如果只用上面的方法，实现出来的只是一个同步的观察者模式。观察者模式本身的目的就是『后台处理，前台回调』的异步机制，因此异步对于 RxJava 是至关重要的。而要实现异步，则需要用到 RxJava 的另一个概念： Scheduler 。

在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler（调度器）。

在RxJava 中，Scheduler ——调度器，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler ，它们已经适合大多数的使用场景：

Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
Schedulers.newThread(): 启用新线程，并在新线程执行操作。
Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

有了这几个 Scheduler ，就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。

subscribeOn(): 指定 subscribe() 所发生的线程，即 ObservableOnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。

文字叙述总归难理解，上代码：

Observable.just(1, 2, 3, 4)
    .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
    .subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer number) {
            Log.d(tag, "number:" + number);
        }
    });

上面这段代码中，由于 subscribeOn(Schedulers.io()) 的指定，被创建的事件的内容 1、2、3、4 将会在 IO 线程发出；而由于 observeOn(AndroidScheculers.mainThread()) 的指定，因此 subscriber 数字的打印将发生在主线程。事实上，这种在 subscribe() 之前写上两句 subscribeOn(Scheduler.io())和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常见，它适用于多数的『后台线程取数据，主线程显示』的程序策略。

5. 操作符

5.1 map

map 可以将一个对象转换为另外一个对象

Observable.just("123").map(Integer::parseInt)
    .subscribe(System.out::println);

5.2 flatMap

举个例子：一个学生有一个名字，有多个课程，现在给你一个学生，输出学生的课程

Observable.just(student)
        .flatMap(stu -> stu.getCourses())
        .subscribe(System.out::println);

从上面的代码可以看出， flatMap() 和 map() 有一个相同点：它也是把传入的参数转化之后返回另一个对象。但需要注意，和 map() 不同的是， flatMap() 中返回的是个 Observable 对象，并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。

flatMap() 的原理是这样的：

使用传入的事件对象创建一个 Observable 对象；
并不发送这个 Observable, 而是将它激活，于是它开始发送事件；
每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。而这个『铺平』就是 flatMap() 所谓的 flat。

5.3 防抖

throttleFirst(): 在每次事件触发后的一定时间间隔内丢弃新的事件。常用作去抖动过滤，例如按钮的点击监听器：

RxView.clickEvents(button) // RxBinding 代码，后面的文章有解释   
    .throttleFirst(500, TimeUnit.MILLISECONDS) // 设置防抖间隔为 500ms   
    .subscribe(subscriber);

5.4 doOnSubscribe

此方法在 subscribe 时执行，在 onNext() 前执行。

Observable.just(student)
        .subscribeOn(Schedulers.io())
    	// 在 io 线程执行
        .doOnSubscribe(System.out::println)
        .subscribeOn(AndroidSchedulers.mainThread())
        .observeOn(AndroidSchedulers.mainThread())
    	// 在 main 线程执行
        .subscribe(subscriber);

5.5 filter

query("王")
    .flatMap(list -> Observable.from(list)) 
    .map(student->return student.getGrade())
    .filter(grade->grade>80)
    .subscribe(grade->Log.i(TAG,grade+"");

5.6 take

take 操作符的功能是限定个数，比如这边的功能就是限定我最多需要5个成绩。

query("王")
    .flatMap(list -> Observable.from(list)) 
    .Map(student->return student.getGrade())
    .filter(grade->grade>80)
    .take(5)
    .subscribe(grade->Log.i(TAG,grade+"");

5.7 doOnNext

doOnNext() 允许我们在每次输出一个元素之前做一些额外的事情

query("王")
    .flatMap(list -> Observable.from(list)) 
    .Map(student->return student.getGrade())
    .filter(grade->grade>80)
    .take(5)
    .doOnNext(grade->save(grade))
    .subscribe(grade->Log.i(TAG,grade+"");

6. 使用场景

6.1 结合 Retrofit

使用传统方式

使用 Retrofit 的传统 API，你可以用这样的方式来定义请求：

@GET("/user")
public void getUser(@Query("userId") String userId, Callback<User> callback);

在程序的构建过程中， Retrofit 会把自动把方法实现并生成代码，然后开发者就可以利用下面的方法来获取特定用户并处理响应：

getUser(userId, new Callback<User>() {
    @Override
    public void success(User user) {
        userView.setUser(user);
    }
    
    @Override
    public void failure(RetrofitError error) {
        // Error handling
        ...
    }
};

使用 RxJava 方式

定义同样的请求是这样的：

@GET("/user")
public Observable<User> getUser(@Query("userId") String userId);

使用的时候是这样的：

getUser(userId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<User>() {
        @Override
        public void onNext(User user) {
            userView.setUser(user);
        }
        
        @Override
        public void onCompleted() {
        }
        
        @Override
        public void onError(Throwable error) {
            // Error handling
            ...
        }
    });

6.2 结合 RxBindint

RxBinding 是 Jake Wharton 的一个开源库，它提供了一套在 Android 平台上的基于 RxJava 的 Binding API。所谓 Binding，就是类似设置 OnClickListener 、设置 TextWatcher 这样的注册绑定对象的 API。

举个设置点击监听的例子。使用 RxBinding ，可以把事件监听用这样的方法来设置：

RxView.clickEvents(button)
    // 去抖动
    .throttleFirst(500, TimeUnit.MILLISECONDS)
    .subscribe(clickAction);

如果觉得文章对你有用，请随意赞赏

java RxJava

RxJava

https://blog.rainsheep.cn/archives/1624351544028

作者

Rainsheep

发布于

2021-06-22

更新于

2023-12-28

许可协议

CC BY 4.0