本篇文章是阅读官方文档的笔记。
作者：shixinzhang（百度搜索 “shixinzhang CSDN” 即可找到我）

RxJava 也用了有段时间，那么多操作符总不想去记，遇到了才去查一下，查的次数多了感觉效率太低，还不如先过一遍官方文档。

因此接下来几篇我们一起刷一遍 RxJava 官方文档，这个过程可能会很枯燥，但是像电影里少林寺练功一样，只有先通过枯燥的学习掌握基本功，然后才能考虑如何应用，加油！

RxJava 1.x 笔记：创建型操作符

读完本文你将了解：

创建型操作符
- create
- defer
- Empty
- Never
- Throw
- From
- Interval
- Just
- Range
- Repeat
- RepeatWhen
- Start
- Timer
Thanks

创建型操作符

创建型操作符即用于创建 Observable 的操作符。

create

RxJava 1.x 笔记：创建型操作符

RxJava 中，create() 通过一个传递一个 Observable.OnSubscribe 创建一个Observable，Observable.OnSubscribe 的泛型参数指明要创建的数据类型：

//使用 create 方法创造被观察者mObservable = Observable.create(new Observable.OnSubscribe<String>() {    /** * 当 observable 被订阅时，会自动调用 call() 方法，依次触发其中的事件 * 其实就是调用订阅者的回调方法，即实现了被观察者向观察者的事件传递 * @param subscriber */    @Override    public void call(Subscriber<? super String> subscriber) {        if (!subscriber.isUnsubscribed()) {            //这里可能会有复杂的事件发送规则            for (int i = 0; i < 5; i ++) {                subscriber.onNext("next 1");            }            subscriber.onCompleted();   //必须得调用 completed        }        //意外情况下，调用 onError        subscriber.onError(new Throwable("Error message"));    }});

在 create() 中根据创建任务的状态分别调用 onNext(), onCompleted() 和 onError()。

在 create() 中最好调用 isUnsubscribed() 检查观察者的订阅状态，这样在没有观察者时可以避免做无用的创建工作。

create() 默认不在任何特定的调度器上执行。

defer

RxJava 1.x 笔记：创建型操作符

defer 中文意思 “推迟、延迟”。

defer 操作符，只有观察者订阅后才会使用一个 Observable 工厂方法创建 Observable ，每次有新的观察者订阅时，都会重复这个操作。

订阅者以为订阅的是同一个数据源，其实是各自订阅的是不同的 Observable。

RxJava 中对应的实现是 defer() ，接受一个 Func0<Observable<T>> 类型的参数：

public final static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {    return create(new OnSubscribeDefer<T>(observableFactory));}

我们可以传入一个 Func0<Observable<T>> 的匿名内部类，泛型参数是数据类型：

mObservable = Observable.defer(new Func0<Observable<String>>() {    @Override    public Observable<String> call() {        return "test";    }});

Empty

Empty 用于创建一个不发射任何数据，但是会发射终止消息的 Observable。

RxJava 1.x 笔记：创建型操作符

RxJava 中对应的实现是 empty() 方法：

public final static <T> Observable<T> empty() {    return (Observable<T>) EmptyHolder.INSTANCE;}//懒加载、静态内部类单例，只发射一个 onCompleted()private static final class EmptyHolder {    final static Observable<Object> INSTANCE = create(new OnSubscribe<Object>() {        @Override        public void call(Subscriber<? super Object> subscriber) {            subscriber.onCompleted();        }    });}

使用起来也很简单，没有参数：

//创建一个不发射任何数据，只会发射 onCompleted()mObservable = Observable.empty();

Never

Never 用于创建一个不发射任何数据，也不会发射终止消息的 Observable （意义何在？）。

RxJava 1.x 笔记：创建型操作符

RxJava 中对应的实现是 never() 方法：

public final static <T> Observable<T> never() {    return NeverObservable.instance();}private static class NeverObservable<T> extends Observable<T> {    private static class Holder {        static final NeverObservable<?> INSTANCE = new NeverObservable<Object>();    }    @SuppressWarnings("unchecked")    static <T> NeverObservable<T> instance() {        return (NeverObservable<T>) Holder.INSTANCE;    }    NeverObservable() {        super(new OnSubscribe<T>() {            @Override            public void call(Subscriber<? super T> observer) {                // do nothing            }        });    }}

可以看到这个方法返回的 NeverObservable 不会发出任何消息，使用起来和 empty() 类似：

//不会发出任何消息mObservable = Observable.never();

Throw

Throw 用于创建一个不发射任何数据，但是会发射错误消息的 Observable。

RxJava 1.x 笔记：创建型操作符

RxJava 中对应的实现是 error() 方法：

public final static <T> Observable<T> error(Throwable exception) {    return new ThrowObservable<T>(exception);}private static class ThrowObservable<T> extends Observable<T> {    public ThrowObservable(final Throwable exception) {        super(new OnSubscribe<T>() {            @Override            public void call(Subscriber<? super T> observer) {                observer.onError(exception);            }        });    }}

可以看到，error() 会发射一个 onError() 事件。

使用很简单，只需传入一个 Throwable：

//只会发出错误消息mObservable = Observable.error(new Throwable("error message"));

Empty/Never/Throw 三个操作符创建的 Observable 比较特殊，一般用于测试，有时也结合其它的数据源，作为其它操作符的参数。

From

From 可以将其他对象或者数据类型转换成 Observables。

RxJava 1.x 笔记：创建型操作符

当使用 Observable 时，最好是所有数据源都是 Observable 类型，那样配合操作符将十分方便。这时如果有 Observable 以外的其他数据类型，我们最好使用一些操作符将其他类型转换为 Observable 。

比如 Iterable 迭代器，可以转换成一系列同步的 Observable；Future，可以转做每次只发射一个元素的 Observable。通过显式地将其他类型转换成 Observable，就可以方便地体会 Rx 的便捷之处。

大部分 ReactiveX 的实现语言都提供了将特定的对象和数据结构转换为 Observables 的方法。

RxJava 中对应的实现是 from():

public final static <T> Observable<T> from(T[] array) {    return from(Arrays.asList(array));}public final static <T> Observable<T> from(Iterable<? extends T> iterable) {    return create(new OnSubscribeFromIterable<T>(iterable));}public static <T> Observable<T> from(Future<? extends T> future) {    return (Observable<T>)unsafeCreate(OnSubscribeToObservableFuture.toObservableFuture(future));}

from() 支持参数为数组、Iterable、Future，也就是说，这个操作符可以将数组、Iterable 和 Future 类型的数据转换为 Observable。

对于 Iterable 和数组，转换后的 Observable 会发射 Iterable 或数组中的每一项数据。

不了解 Iterable 的同学可以看这篇：Java 集合源码解析：Iterator。

使用例子：

private void createObservables() {    String[] words = {"shixin", "is", "cute"};    Observable<String> from = Observable.from(words);    from.subscribe(new Subscriber<String>() {        @Override        public void onCompleted() {            System.out.println("onCompleted");        }        @Override        public void onError(Throwable e) {        }        @Override        public void onNext(String s) {            System.out.println("onNext: " + s);        }    });}

运行结果：

RxJava 1.x 笔记：创建型操作符

from() 默认不在任何特定的调度器上执行。我们可以将 Scheduler 作为第二个参数传递给Observable，这个 Future 将会在指定的线程执行。

Interval

Interval 用于创建一个按指定时间间隔、发送递增的整数序列的 Observable。

RxJava 1.x 笔记：创建型操作符

Interval 操作符返回的 Observable 会间隔发射一个无限递增的整数序列。

RxJava 中对应的实现为 interval()，它接受一个表示时间间隔的参数和一个表示时间单位的参数：

public static Observable<Long> interval(long interval, TimeUnit unit) {    return interval(interval, interval, unit, Schedulers.computation());}public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {    return unsafeCreate(new OnSubscribeTimerPeriodically(initialDelay, period, unit, scheduler));}

使用例子：

private void createObservableWithInterval() {    Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);    interval.subscribe(getPrintSubscriber());}/** * 用于打印结果的订阅者 * @param <T> * @return */private <T> Subscriber<T> getPrintSubscriber() {    return new Subscriber<T>() {        @Override        public void onCompleted() {            System.out.println("onCompleted");        }        @Override        public void onError(Throwable e) {            System.out.println("onError: " + e.getMessage());        }        @Override        public void onNext(T t) {            System.out.println("onNext: " + t);            if ((Long)t > 10){                unsubscribe();            }        }    };}

在这个例子中我们使用 interval() 创建了一个 Observable，它将发射递增的整数序列。在 onNext() 中我们当 t > 10 时取消订阅。运行结果：

RxJava 1.x 笔记：创建型操作符

Just

Just 用于创建一个发射特定元素的 Observable。

RxJava 1.x 笔记：创建型操作符

Just 操作符用于将一个元素转换成 Observable，然后将它发射出去。

Just 和 From 有些相似，不同之处在于 From 会将数组或 Iterable 的数据取出后逐个发射；而 Just 只是简单的原样发射，将数组或 Iterable 当做单个数据一次性发射出去。

RxJava 中对应的实现是 just() ：

public static <T> Observable<T> just(final T value) {    return ScalarSynchronousObservable.create(value);}public static <T> Observable<T> just(T t1, T t2) {    return from((T[])new Object[] { t1, t2 });}

可以看到，多个参数的 just() 最后其实还是调用的 from()。

使用例子：

private void createObservableWithJust() {    String[] words = {"shixin", "is", "cute"};    Observable<String[]> just = Observable.just(words);    just.subscribe(getPrintSubscriber());}

我们传递给 just() 一个数组，结果输出的还是数组对象；传递给一个 null，输出的也是 null：

RxJava 1.x 笔记：创建型操作符

可以看到，just() 果然和名字一样，就是将转换的数据直接发出去。

注意，如果你传递 null 给 just，它会返回一个 Observable 然后发射出去一个 null，而不会创建一个空的 Observable。如果想创建一个空的，你需要使用 Empty 操作符。

Range

Range 用于创建一个发射指定范围的整数序列的 Observable。

RxJava 1.x 笔记：创建型操作符

RxJava 中对应的实现为 range():

public static Observable<Integer> range(int start, int count) {    if (count < 0) {        throw new IllegalArgumentException("Count can not be negative");    }    if (count == 0) {        return Observable.empty();    }    if (start > Integer.MAX_VALUE - count + 1) {        throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");    }    if (count == 1) {        return Observable.just(start);    }    return Observable.unsafeCreate(new OnSubscribeRange(start, start + (count - 1)));}

可以看到，just() 接受两个参数，第一个是起始值，第二个是个数。如果你将第二个值设置为 0，就不会发射数据；如果设置为负数或者起始值太大，就会抛出异常。

使用例子：

private void createObservableWithRange() {    Observable<Integer> range = Observable.range(3, 5);    range.subscribe(this.<Integer>getPrintSubscriber());}

运行结果：

RxJava 1.x 笔记：创建型操作符

Repeat

Repeat 用于创建一个多次发射特定数据的 Observable。

RxJava 1.x 笔记：创建型操作符

RxJava 中的实现 repeat() 不是静态的，也就是说它不可以用于创建 Observable，只可以对已有的 Observable 进行重复发射，参数指定重复次数。

public final Observable<T> repeat(final long count) {    return OnSubscribeRedo.<T>repeat(this, count);}

使用时需要先创建一个 Observable，然后对他进行 repeat 操作：

private void createObservableWithRepeat() {    String[] words = {"shixin", "is", "cute"};    Observable<String> from = Observable.from(words);    from.repeat(2)            .subscribe(getPrintSubscriber());}

运行结果：

RxJava 1.x 笔记：创建型操作符

可以看到，当 repeat() 接收到 onCompleted() 事件后触发重订阅。

repeat 操作符默认在 trampoline 调度器上执行。我们可以自行指定 Scheduler。

RepeatWhen

RepeatWhen 相对费解一些。

Repeat 接收到 onCompleted() 事件后就会触发重订阅，而 RepeatWhen 则在它的基础上，增加了什么时候重订阅的控制。

RxJava 1.x 笔记：创建型操作符

RepeatWhen 在接受到 onCompleted 事件后，会进行条件检查，然后进行相应的重发操作；在接收到 onError 事件后，会停止重复。

RxJava 中对应的实现是 repeatWhen():

public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler) {    return OnSubscribeRedo.repeat(this, InternalObservableUtils.createRepeatDematerializer(notificationHandler));}

可以看到， repeatWhen() 的第一个参数是输入的 Observable 的处理函数，第二个参数是处理策略。

由于只有在调用 onCompleted 时才会调用参数，所以它是 Void 类型。

使用例子：

Observable.unsafeCreate(new Observable.OnSubscribe<String>() {    @Override    public void call(final Subscriber<? super String> subscriber) {        for (int i = 0; i < 5; i++) {                subscriber.onNext("item " + i);        }        subscriber.onCompleted();    }}).repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {    @Override    public Observable<?> call(final Observable<? extends Void> completed) {        //每次调用 onCompleted，都会进入这里，需要在这里决定是否需要重订阅        return completed.delay(5, TimeUnit.SECONDS);    }}).subscribe(getPrintSubscriber());

每次发射 onCompleted 事件时会进入 repeatWhen 的匿名内部类参数，在其中我们做了延迟处理。

运行结果：

RxJava 1.x 笔记：创建型操作符

可以看到，每过 5 秒重复一次。

Start

Start 的作用是创建一个发射函数返回值的 Observable。

RxJava 1.x 笔记：创建型操作符

编程语言中有很多种获取计算结果的方法，比如 functions, futures, actions, callables, runnables 等等。Start 操作符可以将他们的结果转为 Observable，从而更方便地进行操作。

RxJava 中 Start 的实现有很多种，它们都属于 rxjava-async 模块。

在 RxJava 中 From 操作符可以将 Future 转换为 Observable，与 start 相似。

Timer

Timer 用于创建一个在指定延迟后发射值的 Observable。

RxJava 1.x 笔记：创建型操作符

RxJava 中对应的实现是 timer:

public static Observable<Long> timer(long delay, TimeUnit unit) {    return timer(delay, unit, Schedulers.computation());}public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {    return unsafeCreate(new OnSubscribeTimerOnce(delay, unit, scheduler));}

timer() 返回一个Observable，它会在延迟给定的时间后发射一个数字 0。

private void createObservableWithTimer() {    Observable.timer(3, TimeUnit.SECONDS)            .subscribe(this.<Long>getPrintSubscriber());}

运行结果：

RxJava 1.x 笔记：创建型操作符

我们走完了约五分之一的流程，坚持就是胜利！！！

Thanks

http://reactivex.io/documentation/operators.html
http://www.jianshu.com/p/023a5f60e6d0

秒客网

RxJava 1.x 笔记：创建型操作符

创建型操作符

create

defer

Empty

Never

Throw

From

Interval

Just

Range

Repeat

RepeatWhen

Start

Timer

Thanks

相关文章