创建可观察对象：rxjava2的创建操作符详解

# 1. 引言

## 1.1 什么是可观察者模式

可观察者模式是一种软件设计模式，也被称为发布-订阅模式。在这种模式中，存在一个被观察的主题对象，以及多个依赖该主题对象的观察者对象。当主题对象的状态发生变化时，它会通知所有的观察者对象，并且观察者对象会自动更新。

在软件开发中，可观察者模式被广泛应用于事件驱动的编程模型中，比如前端开发中的UI事件处理、后端开发中的消息通知等场景。

## 1.2 RxJava2简介

RxJava是一个在Java虚拟机上使用可观察序列来组成异步和基于事件的程序的库。RxJava2是RxJava的一个重大升级版本，提供了优化性能和许多新功能。

RxJava2基于观察者模式的思想，提供了丰富的操作符和调度器，使得在处理异步事件和数据流时变得更加简单和高效。它的应用广泛，涵盖了Android、后端服务等多个领域。

在接下来的章节中，我们将重点介绍RxJava2中的创建操作符，并通过示例详细说明它们的基本用法和实际应用。

# 2. 创建操作符概览

在RxJava2中，创建操作符用于创建可观察者对象，它们允许我们通过各种方式来创建数据序列。创建操作符可以将数据源转换为可观察者对象，并提供用于订阅的观察者。通过使用不同的创建操作符，我们可以满足不同的需求。

#### 2.1 创建操作符的作用

创建操作符主要用于以下几个方面：

- 将已有的数据序列转换为可观察者对象
- 从数据库、文件、网络等异步接口获取数据并转换为可观察者对象
- 定时生成一系列的数据并转换为可观察者对象

#### 2.2 RxJava2中常用的创建操作符介绍

RxJava2提供了多种创建操作符，常用的包括：

- just(): 将一个或多个对象转换为发射这些对象的可观察者
- from(): 将一个Iterable、数组或Future对象转换为可观察者
- interval(): 定时生成一个从0开始递增的长整型序列
- create(): 创建一个自定义的可观察者

接下来，我们将详细介绍这些创建操作符的使用方法和示例。

# 3. 创建操作符详解之just()

在RxJava2中，创建操作符用于创建一个发射指定数据项的Observable。其中，`just()`是最简单和常用的创建操作符之一。

## 3.1 just()的基本用法

`just()`操作符可以接收一个或多个参数，并将其转换为发射这些参数的Observable。它会依次发射参数中的数据项，并在发射完所有数据项后终止。

下面是`just()`的基本语法：

public static <T> Observable<T> just(T... items)

`T`表示数据项的类型，`items`为可变参数，可以传入任意个数的数据项。

## 3.2 just()的使用示例

Observable<String> observable = Observable.just("Hello", "World");
observable.subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
        // TODO: 在订阅时触发
    }

    @Override
    public void onNext(String s) {
        // TODO: 在接收到数据项时触发，此处会依次接收到"Hello"和"World"
    }

    @Override
    public void onError(Throwable e) {
        // TODO: 在发生错误时触发
    }

    @Override
    public void onComplete() {
        // TODO: 在完成数据项的发射后触发
    }
});

上述示例中，创建了一个Observable对象`observable`，使用`just()`操作符发射了两个字符串数据项"Hello"和"World"。然后，通过`subscribe()`方法订阅这个Observable，并实现了相应的观察者方法。在`onNext()`方法中，我们可以依次接收到"Hello"和"World"两个数据项。

总结：`just()`操作符可以将指定的数据项转换为一个发射这些数据项的Observable对象。它适用于只需要少量数据项的场景，且无法动态添加或删除数据项。

# 4. 创建操作符详解之from()

#### 4.1 from()的基本用法

`from()`操作符是RxJava中的一个创建操作符，它可以将一个数组或Iterable对象转换为Observable对象。通过`from()`操作符，我们可以将一组数据逐个发射出去。

`from()`操作符的基本用法如下：

public static <T> Observable<T> fromArray(T[] array)
public static <T> Observable<T> fromIterable(Iterable<? extends T> iterable)

其中，`fromArray()`方法接收一个数组作为参数，而`fromIterable()`方法接收一个Iterable对象作为参数。这两个方法都会返回一个发射指定数据序列的Observable对象。

#### 4.2 from()的使用示例

下面是一个使用`from()`操作符的示例：

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

import java.util.Arrays;
import java.util.List;

public class FromExample {
    public static void main(String[] args) {
        // 使用fromArray()方法将数组转换为Observable对象
        Integer[] numbers = new Integer[]{1, 2, 3, 4, 5};
        Observable<Integer> fromArrayObservable = Observable.fromArray(numbers);

        // 使用fromIterable()方法将Iterable对象转换为Observable对象
        List<String> fruits = Arrays.asList("Apple", "Banana", "Orange");
        Observable<String> fromIterableObservable = Observable.fromIterable(fruits);

        // 订阅Observable对象，并打印每个发射的数据
        Disposable disposable1 = fromArrayObservable.subscribe(number -> System.out.println(number));
        Disposable disposable2 = fromIterableObservable.subscribe(fruit -> System.out.println(fruit));

        // 取消订阅
        disposable1.dispose();
        disposable2.dispose();
    }
}

代码解析：
- 首先，我们创建了一个数组`numbers`和一个List对象`fruits`作为数据源。
- 然后，使用`fromArray()`方法将数组转换为Observable对象`fromArrayObservable`，使用`fromIterable()`方法将List对象转换为Observable对象`fromIterableObservable`。
- 接下来，我们分别订阅了这两个Observable对象，并在订阅时传入了一个Lambda表达式作为处理数据的方式。对于`fromArrayObservable`，我们打印了每个发射的数字；对于`fromIterableObservable`，我们打印了每个发射的水果名称。
- 最后，我们调用了`dispose()`方法取消订阅。

运行以上代码，输出如下：

1
2
3
4
5
Apple
Banana
Orange

上述示例中，我们通过`from()`操作符将一个数组和一个Iterable对象转换为了Observable对象，并对Observable对象进行了订阅。通过订阅，我们可以按顺序打印出数组和List中的每个元素。

在实际应用中，`from()`操作符可以帮助我们将一组数据转换成Observable对象，便于使用RxJava进行后续的操作和处理。

# 5. 创建操作符详解之interval()

### 5.1 interval()的基本用法

在RxJava2中，`interval()`操作符用于创建一个定时发射整数序列的Observable。它会每隔一段时间发射一个递增的整数，并且可以指定起始延迟时间、间隔时间和时间单位。

`interval()`操作符的基本用法如下：

public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit)

参数说明：
- `initialDelay`：起始延迟时间，即第一次发射的延迟时间。
- `period`：间隔时间，即发射的整数序列的间隔时间。
- `unit`：时间单位，如`TimeUnit.MILLISECONDS`代表毫秒，`TimeUnit.SECONDS`代表秒，以此类推。

### 5.2 interval()的使用示例

下面是一个使用`interval()`操作符创建Observable的示例：

import io.reactivex.rxjava2.core.Observable;
import java.util.concurrent.TimeUnit;

public class IntervalExample {

    public static void main(String[] args) throws InterruptedException {
        Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS);
        observable.subscribe(
                value -> System.out.println("Received: " + value),
                error -> System.out.println("Error: " + error),
                () -> System.out.println("Completed")
        );
        Thread.sleep(5000);
    }
}

代码解析：
- 首先，我们使用`interval()`创建了一个Observable，起始延迟时间为1秒，间隔时间为1秒；
- 然后，我们通过`subscribe()`方法订阅了这个Observable，并设置了对应的 onNext、onError 和 onComplete 回调；
- 最后，我们让主线程休眠5秒钟，以便观察到Observable的发射行为。

运行结果：

Received: 0
Received: 1
Received: 2
Received: 3
Received: 4
Completed

可以看到，Observable每隔1秒发射一个递增的整数，共发射了5个整数后，自动调用了 onComplete 回调。

在实际的应用中，`interval()`操作符可以用来定时执行某些任务，例如定时获取服务器数据、定时刷新UI等。

# 6. 创建操作符详解之create()

在RxJava2中，我们可以使用`create()`操作符来自定义一个可观察者。通过`create()`操作符，我们可以手动定义发射事件的逻辑，并且在其中调用`onNext()`、`onError()`、`onComplete()`等方法。

#### 6.1 create()的基本用法

`create()`方法接收一个`ObservableOnSubscribe`对象作为参数，`ObservableOnSubscribe`是一个函数式接口，用于定义发射事件的逻辑。在`ObservableOnSubscribe`接口中，我们需要手动调用观察者的`onNext()`、`onError()`、`onComplete()`等方法来发射事件。

具体的使用示例如下：

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        // 发射数据
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onNext(3);
        // 发射数据完成
        emitter.onComplete();
    }
}).subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        // 进行一些初始化操作
    }

    @Override
    public void onNext(Integer integer) {
        // 处理接收到的数据
        System.out.println("接收到数据：" + integer);
    }

    @Override
    public void onError(Throwable e) {
        // 错误处理
    }

    @Override
    public void onComplete() {
        // 数据接收完成
    }
});

#### 6.2 create()的使用示例

下面我们通过一个具体的示例来演示`create()`操作符的使用，示例中我们通过`create()`手动发射了一组数据，并在订阅时对数据进行处理：

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        // 模拟发射数据
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onNext(3);
        emitter.onComplete();
    }
}).subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        System.out.println("开始订阅");
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("接收到数据：" + integer);
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("发生错误：" + e.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("数据接收完成");
    }
});

在上面的示例中，通过`create()`操作符手动发射了数据1、2、3，然后在订阅时依次接收并处理了这些数据，最终输出了相应的结果。

通过以上示例，我们可以看到`create()`操作符的基本用法以及如何手动定义发射数据的逻辑。