一、RxJava 基本用法

本章节涉及到的 RxJava 组成要素 :

• Observable（被观察者）： Observable 是一个 可以发送消息的数据源 , 可以同时发送若干消息 , 消息的格式可以通过泛型进行定义 ; 消息发送完毕后 会 通知观察者。Observable 通过 订阅观察者 来实现 消息的传递。

• Observer（观察者）： Observer 表示一个接收 Observable 发送消息 的观察者。 它可以处理从 Observable 发射的消息， 还可以处理错误和完成的事件。

• Subscription（订阅）： 订阅是 Observer 对 Observable 的绑定， 它表示观察者正在接收 Observable 的数据项。 订阅可以被取消， 取消订阅后 Observer 观察者将不再接收 Observable 被观察者 的消息。

1、引入 RxJava 依赖

Gradle 项目中 , 在 build.gradle 构建脚本中 , 添加如下依赖 ;

dependencies {
    implementation 'io.reactivex.rxjava2:rxjava:2.2.21'
}

rxjava3 依赖 :

dependencies {
	implementation 'io.reactivex.rxjava3:rxjava:3.0.0'
}

Maven 项目中 , 在 pom.xml 中添加如下依赖 ;

<dependency>
    <groupId>io.reactivex.rxjava2</groupId>
    <artifactId>rxjava</artifactId>
    <version>2.2.21</version>
</dependency>

rxjava3 依赖 :

<dependency>
    <groupId>io.reactivex.rxjava3</groupId>
    <artifactId>rxjava</artifactId>
    <version>3.0.0</version>
</dependency>

2、定义 Observer 观察者

Observer（观察者）： Observer 表示一个接收 Observable 发送消息 的观察者。 它可以处理从 Observable 发射的消息， 还可以处理错误和完成的事件。

Observer 观察者 是 操作的核心 , 定义在需要进行具体操作的位置 , 执行具体的 异步操作 或 事件 ;

如 : 在 UI 界面中 , 点击按钮 , 查询远程数据库服务器中的数据 , 查询完毕后更新 UI 界面 ;

• 该 Observer 观察者 就需要 定义在 UI 界面中 , 可以获取到相关的 UI 组件进行数据更新 ;
• Observable 被观察者可以定义在 Observer 观察者位置 , 也可以定义在消息发送的位置 , 这里 推荐定义在消息发送的位置 ;
• 调用时 , 将 Observer 观察者 传递给对应的异步操作函数 ;
• 在异步操作函数中 , 创建 Observable 被观察者 , 并且通过订阅将观察者订阅到被观察者中 ;
• 订阅操作 , 就会同时发送消息给 观察者 ;

Observer 观察者定义代码 :

Observer<String> observer = new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
        // 当观察者订阅时的回调
    }

    @Override
    public void onNext(String value) {
        // 当接收到新的事件时的回调
        System.out.println(value);
    }

    @Override
    public void onError(Throwable e) {
        // 当发生错误时的回调
    }

    @Override
    public void onComplete() {
        // 当事件流结束时的回调
    }
};

3、定义 Observable 被观察者

Observable（被观察者）： Observable 是一个 可以发送消息的数据源 , 可以同时发送若干消息 , 消息的格式可以通过泛型进行定义 ; 消息发送完毕后 会 通知观察者。Observable 通过 订阅观察者 来实现 消息的传递。

Observable<String> observable = Observable.just("Hello", "World");

4、Observable 被观察者订阅 Observer 观察者

Subscription（订阅）： 订阅是 Observer 对 Observable 的绑定， 它表示观察者正在接收 Observable 的数据项。 订阅可以被取消， 取消订阅后 Observer 观察者将不再接收 Observable 被观察者 的消息。

调用 Observable 被观察者 的 subscribe 函数 , 订阅 Observer 观察者 ;

该订阅操作的同时 , 会将消息发送给 Observer 观察者 , 触发 Observer#onNext 函数 ;

observable.subscribe(observer);

二、代码示例

代码示例 :

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class Main {
    public static void main(String[] args) {
        Observable<String> observable = Observable.just("Hello", "World", "RxJava");

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                // 当观察者订阅时的回调
            }

            @Override
            public void onNext(String value) {
                // 当接收到新的事件时的回调
                System.out.println(value);
            }

            @Override
            public void onError(Throwable e) {
                // 当发生错误时的回调
            }

            @Override
            public void onComplete() {
                // 当事件流结束时的回调
            }
        };

        observable.subscribe(observer);
    }
}

执行结果 :

Hello
World
RxJava