RxJava2进阶教程（一）——创建型操作符

写在前面的话：RxJava的强大之处不仅在于它的设计思想，更是因为有着各种强大的操作符，操作符让你可以灵活的处理变换、组合、操纵和处理Observable发射的数据，接下来的几篇文章会按照操作符的分类逐步讲解各个操作符的作用，合理使用这些操作符也许你会达到事半功倍的效果。
本文适用于那些RxJava的初学者进阶使用，但是如果你还不知道什么是RxJava，那么我只能说

这篇的主题是创建型操作符，也就是创建Observable的操作符，先来看看都有那些操作符吧：

1. create()
2. from()
3. just()
4. interval()
5. timer()
6. empty()
7. error()
8. range()
9. repeat()
10. defer()
11. never()

1.Create

这个就没什么好说的了吧？大多数时候我们都是用这个函数来创建Observable的，
经典的流试调用

Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            if (!e.isDisposed()) {
                e.onNext(1);
            }
        }
    }).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
        }
    });

2.from

** from的作用是将一个Iterable, 一个Future, 或者一个数组转换成一个Observable**，from方法具体有一下几种：

常用的有frmoArray和fromIterable这两个，来看个具体例子吧
集合遍历

  List<Integer> ls = new ArrayList<>();
    ls.add(1);
    ls.add(2);
    ls.add(3);
    Observable.fromIterable(ls).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            Log.i(TAG, "accept: " + integer);
        }
    });

其他的方法大家自行体会吧

3.just

just是将一个或多个对象转换成发射这个或这些对象的一个Observable，在某些情况下它其实和from的作用有些类似，或者你也可以理解为这是一个简版的create方法。来看具体例子：

Observable.just(1).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            Log.i(TAG, "accept: "+integer);
        }
    });

just可以传入一个或多个对象，然后一次发射出来，对于数据量较小时可以使用这个方法。

4.interval

interval可以创建一个按照给定的时间间隔发射整数序列的Observable，通常我们可以用它来来完成一些周期行的操作，比如心跳，Handler最好的替代品了，来看具体例子：

 Observable.interval(2, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {
        @Override
        public void accept(@NonNull Long aLong) throws Exception {
            Log.i(TAG, "accept: "+aLong);
        }
    });

第一个参数为时间间隔，第二个为时间单位，上述accept()会2S执行一次，在这里你就可以完成一些周期性任务了。

5.timer

timer可以创建一个在给定的延时之后发射单个数据的Observable，通常可以用它完成一些定时任务，注意它和interval的区别，interval是周期性的（多次），timer是定时的（一次），具体例子：

Observable.timer(5, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {
        @Override
        public void accept(@NonNull Long aLong) throws Exception {
            Log.i(TAG, "accept: "+aLong);
        }
    });

参数和interval一样，5s之后执行accept方法

6.empty

empty可以创建一个什么都不做直接通知完成的Observable。这个操作符的使用范围就比较窄了，举一个应用场景，我们做数据缓存时通常的实现逻辑是先加载本地缓存，如果本地没有或者已经过期那么才请求网络，这里就可以使用empty，假设背地缓存不存在，那么是抛出异常还是返回null呢？显然这两种都不合理，因为如果本地不存在我们需要直接请求网络，一种解决办法就是当为空的时候直接返回Observable.empty(),同时借助switchIfEmpty（observable1,observable2）方法完成后续的网络请求。当然你也可以通过concat和first来完成数据的获取，这是我们后续的文章要说的了。场景说忘了，那到底是不是直接通知完成呢？测试一下就知道了

   Observable<Integer> o1 = Observable.empty();
    o1.subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onNext(Integer integer) {
            Log.i(TAG, "onNext: ");
        }
        @Override
        public void onError(Throwable e) {
            Log.i(TAG, "onError: ");
        }
        @Override
        public void onComplete() {
            Log.i(TAG, "onComplete: ");
        }
    });

果然是只有onComplete执行了。

7.error

error可以创建一个不发射数据以一个错误终止的Observable，需要一个Throwable参数，使用场景还是拿缓存举例，比如内存缓存不存在，网络数据也请求失败了，那么我们就可以调用这个操作符抛出一个Throwable。那么就直接执行到了Observer的onError方法里里。测试一下，上面的例子不变，我们把empty换成error，并随便传入一个Throwable：

Observable<Integer> o1 = Observable.error(new IOException());
    o1.subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onNext(Integer integer) {
            Log.i(TAG, "onNext: ");
        }
        @Override
        public void onError(Throwable e) {
            Log.i(TAG, "onError: ");
        }
        @Override
        public void onComplete() {
            Log.i(TAG, "onComplete: ");
        }
    });

直接执行了onError方法。

8.range

range可以创建一个发射指定范围的整数序列的Observable。直接上例子：

Observable.range(0,5).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            Log.i(TAG, "accept: "+integer);
        }
    });

第一个参数为数列的起始元素，第二个为长度，打印结果为：

需要注意的是如果你将第二个参数设为0，将导致Observable不发射任何数据（如果设置为负数，会抛异常）

9. repeat

repeat可以创建一个重复发射指定数据的Observable，你也可以通过repeat(n)指定重复次数。更多时候我们会使用repeat(n)这个方法，具体例子：

 Observable.just(1).repeat(5).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            Log.i(TAG, "accept: "+integer);
        }
    });

结果为：

10.defer

defer的作用是只有当Observer订阅才创建Observable并且是为每个Observer创建一个新的Observable。
正常情况下：

    int a = 10;
    Observable<Integer> o2 = Observable.just(a);
    a = 20;
    o2.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            Log.i(TAG, "accept= " + integer);
        }
    });

输出为accept=10，但是如果用defer

Observable<Integer> o1 = Observable.defer(new Callable<ObservableSource<Integer>>() {
        @Override
        public ObservableSource<Integer> call() throws Exception {
            return Observable.just(a);
        }
    });
    a = 20;
    o1.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            Log.i(TAG, "accept:= " + integer);
        }
    });

输出为accept=20，大家仔细看一下上边两个例子就理解defer了。

11. never

never可以创建一个不发射数据也不终止的Observable，具体还没有使用过，也没想到应用场景大家知道有这么一个操作符就可以了。

以上就是Observable的常见的创建操作符了，有遗漏的欢迎大家补充，下一篇会介绍常见的变换操作符，敬请期待。