RxJava2最全面、最详细的讲解（一）

落日映苍穹つ 2023-07-14 08:22 18阅读 0赞

# 一、概述 #

提高开发效果，降低维护成本一直是团队追求的宗旨。现在安卓里面RxJava2+Retrofit+OKHttp的网络请求框架是最流行的，几乎一统江湖，配合响应式式编程的RxJava使用也越来越多。如果英文比较好而且有兴趣的可以到官网学习：[ReactiveX的官方网站][ReactiveX]。（源码和其他相关链接在文章最后给出）

### **RxJava到底是什么？** ###

RxJava在GitHub上《[RxJava][]》的自我介绍是：a library for composing asynchronous and event-based programs using observable sequences for the Java VM.(一个在Java VM 上使用可观测的序列来组成异步的，基于事件的程序的库)，有些人可能感到疑惑，其实本质上可以用一词来概括——“**异步**”，它就是一个异步的操作库，而别的定语都基于这之上。

### **RxJava的好处是啥？** ###

同样是异步操作，为什么不用Handler或者AsyncTask等异步机制？

因为：**简洁**

异步操作很关键的一点是简洁性，因为在调度过程比较复杂的情况下，异步代码既难写也难读懂。Android创造的Handler和AsyncTask其实都是为了让异步代码更加简洁，RxJava的优势也是简洁，它的特别之处是：**随着程序代码逻辑的变得越来越复杂，它依然保持简洁。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70][]

# 二、使用详解 #

### 1.基本实现 ###

RxJava最核心的两个东西Observable（被观察者、事件源）和Observer（观察者），Observable发出一系列的事件，Observer处理这些事件。在Observer接收到事件处理之前我们很方便地对结果做出各种拦截处理等。

RxJava的流程图大致如下：

![2019101818052569.png][]

**1).添加依赖库**

在model的build.gradle文件内添加RxJava2依赖库，注意RxJava1与RxJava2依赖不能共存。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 1][]

implementation 'io.reactivex.rxjava2:rxjava:2.0.4'
        implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

**2).三步创建实例**

//1、创建被观察者Observable
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("RxJava：e.onNext== 第一次");
                e.onNext("RxJava：e.onNext== 第二次");
                e.onNext("RxJava：e.onNext== 第三次");
                e.onComplete();
            }
        });
     
        //2、创建观察者Observer
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext == " + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete == ");
            }
        };
     
        //3、订阅(观察者观察被观察者)
        observable.subscribe(observer);

直接运行上面的例子，打印结果如下：

![20191014145436821.png][]

上述例子中，通过new Observer<T>()创建观察者，实现相应的方法，Observable.create(ObservableOnSubscribe)创建被观察者，subscribe()将两者关联起来。

**3).观察者模式**

**RxJava的异步实现，是通过一种拓展的观察者模式来实现的。**

### 什么是观察者模式？ ###

其实就是发布订阅模式，发布者发布信息，订阅者接受信息。没订阅就接收不到信息。

观察者模式面向的需求是:A对象(观察者)对B对象(被观察者)的某种变化高度敏感，需要在B对象变化的一瞬间做出变化。程序的观察者模式，观察者不需要时刻盯着被观察者，而是采用**注册(Regsiter)**或者**订阅(Subscribe)**的方式，告诉被观察者，我是你的某种状态，你的状态在发生变化的时候来通知我。

例如：连载小说和读者的关系，观察者是读者，被观察者是连载小说，读者订阅了连载小说，当连载小说发布新的章节时通知并推送给读者，然后读者阅读新的章节，这就是观察者模式。

### RxJava的观察者模式 ###

RxJava有四个基本概念：Observer(观察者)，Observable(被观察者)，subscribe(订阅)，事件。Observer和Observable通过subscribe()实现订阅关系，从而Observable可以在需要的时候发出事件通知Observer。

*  Observer：     观察者，它决定事件发生时有怎么样的行为；
 *  Observable： 被观察者，它决定什么时候出发事件以及触发什么样的事件；
 *  subscribe：    订阅，将Observer和Observable关联起来。

我们来看看上面的例子的具体步骤：

**1）创建被观察者Observable**

它决定什么时候出发事件以及触发怎么样的事件，通过Observable.create(ObservableOnSubscribe)创建被观察者实例，这是最基本的创造事件序列的方法，并为它定义事件出发规则。

//创建被观察者Observable
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("RxJava：e.onNext== 第一次");
                e.onNext("RxJava：e.onNext== 第二次");
                e.onNext("RxJava：e.onNext== 第三次");
                e.onComplete();
            }
        });

创建一个ObservableOnSubscribe对象并且实现subscribe()方法，设定了事件的内容是String，并返回ObservableEmitter，相当于一个计划表，当Observable被订阅的时候，复写subscribe()方法定义发送的事件，ObservableEmitter是事件发射器，定义并且观向察者发送需要发送的事件，onNext()会被执行三次，最后执行onComplete()方法，这样由被观察者调用观察者回调的方法，实现了被观察者向观察者传递事件。

**2）创建观察者Observer**

它决定事件触发有怎样的行为，定义响应事件的行为，直接new一个Observer观察者实例，实现其中相应的方法：

//创建观察者Observer
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext == " + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete == ");
            }
        };

*  onSubscribe(Disposable d)：  事件订阅成功回调，返回Disposable请求实例，可以通过d.dispose()取消请求；
 *  onNext(T)：                             响应事件的方法，发送事件时，观察者会回调onNext()方法，接收事件数据；
 *  onError()：                               事件队列异常，在处理事件出现异常的时候回触发这个方法，其他事件不会再继续发出；
 *  onComplete()：                       事件队列完结，当不再有新onNext()发出时，需要触发onComplete()方法来作为标志，其他事件不会再继续发出

在一个正确的事件运行队列中，onError()和onComplete()有且仅有一个出现，还是在事件的最后出现，即onError()和onComplete()是互斥的，当一个出现了，另一个就不再回出现。

**3）subscribe()**

订阅，连接Observable(被观察者)和Observer(观察者)。

//订阅(观察者观察被观察者)
        observable.subscribe(observer);

Observable是被观察者，observer是观察者，创建完Observer和Observable之后，通过subscribe()将两者关联起来。

通过该调用，回调观察者的相关方法，从而响应被观察者响应的事件，Observable只生产事件，真正发送事件的是在它订阅的时候，即subscribe()被调用的时候。

另外：被观察者Obaservable的subscribe具有多个重载的方法：

//观察者不对被观察者发送的事件做出响应(但是被观察者还可以继续发送事件)
        public final Disposable subscribe()
     
        //观察者对被观察者发送的任何事件都做出响应
        public final void subscribe(Observer<? super T> observer)
     
        //表示观察者只对被观察者发送的Next事件做出响应
        public final Disposable subscribe(Consumer<? super T> onNext)
     
        //表示观察者只对被观察者发送的Next & Error事件做出响应
        public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
     
        //表示观察者只对被观察者发送的Next & Error & Complete事件做出响应
        public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                                          Action onComplete)
     
        //表示观察者只对被观察者发送的Next & Error & Complete & onSubscribe事件做出响应
        public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                                          Action onComplete, Consumer<? super Disposable> onSubscribe)

### 2.Scheduler调度者 ###

在RxJava默认规则中，事件的发出和消费都是在同一个线程中发生的，那么上面的例子来说，就是一个同步的观察者模式。观察者模式的本省就是后台处理，前台回调的异步机制，因此异步对RxJava来说是至关重要的，异步的实现则需要用到Scheduler调度器来切换线程。

在RxJava中Scheduler(调度器)相当于线程控制器，RxJava通过Scheduler来指定那一部分代码执行在哪一个线程。我们来看看简单的例子：

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("RxJava：e.onNext== 第一次");       
                e.onComplete();
                Log.d(TAG, "subscribe()线程==" + Thread.currentThread().getId());
            }
        }).subscribeOn(Schedulers.io())//指定被观察者subscribe()(发送事件的线程)在IO线程()
                .observeOn(AndroidSchedulers.mainThread());//指定观察者接收响应事件的线程在主线程

log如下：

![20191016170314615.png][]

可以看到主线程id为1，执行事件subscribe()方法的线程id为432，回调事件方法onNext()的线程为1，即在主线程中执行。

由于subscribeOn(Schedulers.io())指定了subscribe()方法发送事件线程在IO线程中执行，observeOn(AndroidSchedulers.mainThread())指定了接收事件在主线程中执行。这种方式非常常见，适用于后台获取数据，前台显示的程序策略。

*  subscribeOn()：      用于指定Observable被观察者subscribe()时所发生的线程，即指定发生事件的线程
 *  observeOn()：         指定`Observer观察者接收&响应事件的线程，即订阅者接收事件的线程`

注意：多次指定发射事件的线程只有第一次指定有效，也就是说多次调用subscribeOn()只有第一次有效，其余的会被忽略；但是多次指定订阅者接收事件的线程是可以的，也就是说每observeOn()一次，接收事件的线程就会切换一次。

RxJava中内置了很多线程项供我们选择：

*  Schedulers.io()：                             代表IO操作的线程，通常用于网络、读写文件等IO密集型的操作。行为模式和new                                                                  Thread()差不多，只是IO的内部是一个无上限的线程池，可重用空闲的线程，更高效                                                                （不要把计算工作放在IO内，可以避免创建不必要的线程）；
 *  AndroidSchedulers.mainThread()：Android的主线程；用于更新UI
 *  Schedulers.newThread()：             总是启用新线程，并在新线程中执行操作；多用于耗时操作
 *  Schedulers.computation()：           代表CPU计算密集型的操作，即不会被IO等操作限制性能的操作。

# 三、常用操作方式 #

RxJava的强大之处，在于它提供类丰富且强悍的操作符，通过使用和组合操作符，你几乎能完成你想完成的任务。下面我们来了解这些操作符的含义和使用方法。

### 1.快速创建 ###

**（1）create****()：是所有创建型操作符的“根”，也就是说其他创建型操作符最后都是通过create()来创建的Observable的，快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。**

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                try{
                    if (!emitter.isDisposed()){
                        emitter.onNext("RxJava：e.onNext== 第一次");
                        emitter.onNext("RxJava：e.onNext== 第二次");
                        emitter.onNext("RxJava：e.onNext== 第三次");
                        emitter.onComplete();
                    }
                }catch (Exception e){
                    emitter.onError(e);
                }
            }
        }).subscribe(new Observer<String>() {
            //默认最先复写onSubscribe()
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext == " + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete == ");
            }
        });

在使用create()操作符时，最好在被观察者的回调函数subscribe()中加上isDisposed()，以便在观察者断开连接的时候不在执行subscribe()函数中的相关逻辑，避免意想不到的错误出现。

打印数据如下：

![20191014145436821.png][]

**（2）****empty()：快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。**

//快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。
    Observable.empty()
               .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "empty：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Object o) {
                Log.e(TAG, "empty：onNext ==" + o.toString());
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "empty：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "empty：onComplete == ");
            }
        });

打印log如下：

![20191017160732285.png][]

**（3）error()：快速创建被观察者对象，仅发送onError()事件，直接通知异常。**

//快速创建被观察者对象，仅发送onError()事件，直接通知异常。
        Observable.error(new Throwable("只回调error"))
                .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "error：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Object o) {
                Log.e(TAG, "error：onNext ==" + o.toString());
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "error：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "error：onComplete == ");
            }
        });

打印log如下：

![201910171608046.png][]

**（4）****never()：快速创建被观察者对象，不发送任何事件。**

//快速创建被观察者对象，不发送任何事件。
        Observable.never()
                  .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "never：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Object o) {
                Log.e(TAG, "never：onNext ==" + o.toString());
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "never：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "never：onComplete == ");
            }
        });

打印log如下：

![20191017160828385.png][]

**（5）Just(****T... items)：快速创建被观察者对象，最多只能发送10个事件。**

*  参数类型：任意类型

Observable.just(1, 2, 3, 4, 5).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "just：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "just：onNext == " + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "just：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "just：onComplete == ");
            }
        });

通过just()创建传入Integer类型的参数构建Observable被观察者，相当于执行了onNext(1)~onNext(5)，通过链式编程订阅观察者。log如下：

![20191016203451256.png][]

**（6）fromArray(****T... items)：快速创建被观察者对象，可发送多个任意子类型的数据。**

*  参数类型：数组，子类型可以为任意类型

//设置需要传入的数组
        String[] strings = {"商品类","非商品类"};
        //传入数组，被观察者创建后会将数组转换成Observable并且发送里面所以的数据
        Observable.fromArray(strings).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "fromArray：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(String s) {
                Log.e(TAG, "fromArray：onNext == " + s);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "fromArray：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "fromArray：onComplete == ");
            }
        });

通过fromArray()创建被观察者对象，传入数组，被观察者创建后会将数组转换成Observable(被观察者)并且发送里面所有的数据。打印数据如下：

![20191016205828520.png][]

**（7）****fromIterable(Iterable<? extends T> source)：快速创建被观察者对象，可发送多个任意子类型的数据。**

*  参数类型：集合，子类型可以为任意类型

//创建集合
        List<Goods> list = new ArrayList();
                    for (int i = 0; i < 3; i++) {
            Goods g = new Goods("名称" + i);
            list.add(g);
        }
        //传入集合，被观察者创建后会将数组转换成Observable并且发送里面所有的数据
                    Observable.fromIterable(list).subscribe(new Observer<Goods>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "fromArray：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Goods goods) {
                Log.e(TAG, "fromArray：onNext == " + goods.getName());
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "fromArray：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "fromArray：onComplete == ");
            }
        });

通过**fromIterable()**创建被观察者Observable对象，传入集合，被观察者创建后会将数组转换成Observable(被观察者)并且发送里面所有的数据。打印数据如下：

![20191016212247432.png][]

### 2.延迟创建 ###

**defer** v. 推迟; 延缓; 展期;

**（1）defer()：直到有Observer观察者订阅时，才会通过Observeable的工厂方法动态创建Observeable，并且发送事件**

每次订阅后都会得到一个刚创建的最新的Observable对象，确保被观察者对象的数据是最新的。

//1.初始化i
        Integer i = 100;
        //2.通过defer()定义被观察者（此时被观察者对象还没创建）
        Observable<Integer> defer = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> call() throws Exception {
                return Observable.just(i);
            }
        });
     
        //3.重新设置i值
        i = 200;
     
        //4.订阅观察者（此时才会调用defer,创建被观察者对象）
            defer.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "defer：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer i) {
                Log.e(TAG, "defer：onNext == " + i);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "defer：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "defer：onComplete == ");
            }
        });

在订阅的时候才开始创建被观察者对象，然后再发送事件，所以**i**的值是最新的数据 **i **= 200；打印log如下：

![20191017114300520.png][]

**（2）timer()：快速创建Observable被观察者对象，延迟指定时间后发送一个类型为Long的事件**

**构造方法：**

timer(long delay, TimeUnit unit)
     timer(long delay, TimeUnit unit, Scheduler scheduler)

*  **delay：                 **延时的时间，类型为Long；
 *  **unit：                   **表示时间单位，有TimeUnit.SECONDS等多种类型；
 *  **scheduler：         **表示调度器，用于指定线程。

发送的事件类型为Long，数值为0，相当于onNext(0);

final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
     
        Log.e(TAG, "timer：当前时间 ==" + dateFormat.format(System.currentTimeMillis()));
        //延时10秒后，发送一个long值为0的事件
        Observable.timer(3, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "timer：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Long aLong) {
                Log.e(TAG, "timer：onNext ==" + aLong + "   时间 ==" + dateFormat.format(System.currentTimeMillis()));
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "timer：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "timer：onComplete == ");
            }
        });

延时3秒后，发送一个long值为0的事件，约3秒后响应事件onNext()接收到数据，另外，timer默认运行在一个新的线程上，也可以自定义线程调度器scheduler。打印log如下：

![20191017122441167.gif][]

**（3）interval()：快速创建Observable被观察者对象，每隔指定的时间就发送相应的事件，事件序列从0开始，无限递增1；**

**interval**  
n.(时间上的)间隔，间隙，间歇;

**构造方法：**

//在指定延迟时间后，每个多少时间发送一次事件
    interval(long initialDelay, long period, TimeUnit unit)
     
    //在指定的延迟时间后，每隔多少时间发送一次事件，可以指定调度器
    interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler)
     
    //每间隔多少时间发送一次事件，使用默认的线程
    Observable<Long> interval(long period, TimeUnit unit)
     
    //每间隔多少时间发送一次事件，可以指定调度器
    interval(long period, TimeUnit unit, Scheduler scheduler)

*  **initialDelay： **表示延迟开始的时间，类型为Long；
 *  **period：         **距离下一次发送事件的时间间隔，类型为Long；
 *  **unit：              **时间单位，有TimeUnit.SECONDS等多种类型；
 *  **scheduler：    **表示调度器，用于指定线程。

interval()提供了多种构造方法选择，每隔一定的时间period就发送一次事件，事件数据从0开始，无限增加1。

//initialDelay：表示延迟开始的时间, period：距离下一次发送事件的时间间隔, unit：时间单位    
     Observable.interval(3, 1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "interval：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Long aLong) {
                Log.e(TAG, "interval：onNext ==" + aLong);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "interval：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "interval：onComplete == ");
            }
        });

上述列子延迟3秒后开始发送事件，每次发送事件的时间间隔为1秒，无限发送，时间单位设置为秒，这种情景我们可以用来设置定时器等相关操作，打印log如下：

![20191017144622123.gif][]

**（4）****intervalRange()：类似于interval()，快速创建一个被观察者对象，指定时间间隔就发送事件，可以执行发送事件的数量，数据依次递增1。**

**构造方法：**

intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit)
     
    intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, Scheduler scheduler)

*  **start：              **表示事件开始的数值大小，类型为Long；
 *  **count：            **表示事件执行的次数，类型为long，不能为负数;
 *  **initialDelay：   **表示延迟开始的时间，类型为Long；
 *  **period：           **距离下一次发送事件的时间间隔，类型为Long；
 *  **unit：                **时间单位，有TimeUnit.SECONDS等多种类型；
 *  **scheduler：     **表示调度器，用于指定线程。

注意：count不能为负数，否则会抛出异常，我将count设置为-1，运行报错如下：

![20191017152452359.png][]

//事件开始大小为10，发送5次事件，延迟3秒后执行，每次执行的间隔为1，单位为秒
    Observable.intervalRange(10, 5, 3, 1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "intervalRange：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Long aLong) {
                Log.e(TAG, "intervalRange：onNext ==" + aLong);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "intervalRange：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "intervalRange：onComplete == ");
            }
        });

这里指定了事件开始大小为10，发送5次事件，延迟3秒后执行，每次执行的间隔为1，单位为秒，打印log如下：

![20191017151410756.gif][]

**（5）range(****final int start, final int count)：类似于intervalRange()，快速创建一个被观察者对象，指定事件起始值，执行发送事件的数量，****但是区别在于range()不能延迟发送的时间。**

注意：上述参数的类型是int类型，rangeLong(long start, long count)只是参数类型不同，用法一致。

//start：事件的开始值大小，count：发送的事件次数
    Observable.range(5, 4).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "range：onSubscribe == 订阅");
            }
     
            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "range：onNext ==" + integer);
            }
     
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "range：onError == " + e.getMessage());
            }
     
            @Override
            public void onComplete() {
                Log.e(TAG, "range：onComplete == ");
            }
        });

这里指定了事件的开始值大小为5，发送的事件数为4次，一次递增1；打印log如下：

![20191017152947800.png][]

上述的几种常用的创建操作符总结一下：

<table> 
 <tbody> 
  <tr> 
   <th colspan="2">创建类型</th> 
   <th>作用</th> 
   <th>备注</th> 
   <th>使用场景</th> 
  </tr> 
  <tr> 
   <td>基本创建</td> 
   <td>create()</td> 
   <td>创建一个完成的被观察者(Observable)</td> 
   <td>RxJava中创建被观察者最基本的操作符</td> 
   <td colspan="1"> <p>1、完整&amp;快速创建被观察者</p> <p>2、数组、集合遍历</p> </td> 
  </tr> 
  <tr> 
   <td colspan="1">快速创建</td> 
   <td>empty()</td> 
   <td>快速创建后只发送complete事件，直接通知完成</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>error()</td> 
   <td>快速创建后只发送error事件，直接通知异常</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>never()</td> 
   <td>快速创建后不发送任何事件</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>just()</td> 
   <td>快速创建后直接发送传入的事件</td> 
   <td>参数最多只能10个，即发送的事件最多10个</td> 
  </tr> 
  <tr> 
   <td>fromArray()</td> 
   <td>快速创建后直接发送传入的数组数据</td> 
   <td>参数为数组，子类型为任意类型，可10个以上</td> 
  </tr> 
  <tr> 
   <td>fromIterable()</td> 
   <td>快速创建后直接发送传入的集合List数据</td> 
   <td>参数为集合List，子类型为任意类型，可10个以上</td> 
  </tr> 
  <tr> 
   <td colspan="1">延时创建</td> 
   <td>defer()</td> 
   <td>直到观察者Observer订阅被观察者Observable时，才动态创建被观察者&amp;发送事件</td> 
   <td>通过Observable工厂方法创建被观察者，每次订阅后都会得到一个最新创建的被观察者Observable，确保里面的数据是最新的</td> 
   <td colspan="1"> <p>1、定时操作</p> <p>2、周期性操作</p> </td> 
  </tr> 
  <tr> 
   <td>timer()</td> 
   <td>快速创建被观察者，指定延时时间，发送一个数值为0的事件</td> 
   <td>延时指定时间后发送一个参数为0的事件，相当于onNext（0）</td> 
  </tr> 
  <tr> 
   <td>interval()</td> 
   <td>快速创建被观察者，每隔指定时间发送事件</td> 
   <td>发送事件序列，从0开始，无限递增1</td> 
  </tr> 
  <tr> 
   <td>intervalRange()</td> 
   <td>快速创建被观察者，每隔指定时间发送事件，可指定发送事件数</td> 
   <td>发送无限递增1的事件序列，可指定起始值大小和事件次数，可延时，类似interval()</td> 
  </tr> 
  <tr> 
   <td>range()</td> 
   <td>快速创建被观察者，连续发送一个时间序列，可指定范围</td> 
   <td>发送无限递增1的事件序列，可指定起始值大小和事件次数，不可延时，类似intervalRange()</td> 
  </tr> 
  <tr> 
   <td>rangeLong()</td> 
   <td>同上，区别在于数据参数为类型Long</td> 
   <td>同上</td> 
  </tr> 
 </tbody> 
</table>

本来想一篇写完的，但是考虑到篇幅太长影响阅读的效率，所以还是另外再开一篇来讲解其他的用法，想继续了解的可以点击[RxJava2最全面、最详细的用法讲解（二）][RxJava2]。

至此，本文结束！

> **源码地址：**[https://github.com/FollowExcellence/Rxjava\_Retrofit][https_github.com_FollowExcellence_Rxjava_Retrofit]

[ReactiveX]: http://reactivex.io/documentation/scheduler.html
[RxJava]: https://github.com/ReactiveX/RxJava
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70]: /images/20230529/91ee85085cdb44578f5f6c94212fb373.png
[2019101818052569.png]: /images/20230529/d3005add12d34c35b0f0fde2d3f4d377.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3Nzk2Njgz_size_16_color_FFFFFF_t_70 1]: /images/20230529/f675cba0902c4147896538c8d2fd5b89.png
[20191014145436821.png]: /images/20230529/9082aba82d924bf192d3db7e86cbcd20.png
[20191016170314615.png]: /images/20230529/3b01efeea28b48a98edd6612baae333e.png
[20191017160732285.png]: /images/20230529/e1cfa568e9974afb8d939e41ba52f64d.png
[201910171608046.png]: /images/20230529/f530ed35c308443391dc255e9e082640.png
[20191017160828385.png]: /images/20230529/3392020f52534cf2aba78e09972c3dec.png
[20191016203451256.png]: /images/20230529/c696352e86c74887b99156f55d03fbea.png
[20191016205828520.png]: /images/20230529/d10100378df24f1887c6a20e1a3b5feb.png
[20191016212247432.png]: /images/20230529/1172e28a6be64c10a360ef51e366728e.png
[20191017114300520.png]: /images/20230529/7bc1c03f02174c20b9cdfdaa10ac59fc.png
[20191017122441167.gif]: /images/20230529/d23aecad1f014eb78ef65ce2a4263b02.png
[20191017144622123.gif]: /images/20230529/c37b137028f04532b25a752776f7ff66.png
[20191017152452359.png]: /images/20230529/da68a8e3f68f4966b0cfb6cbbeee3e6a.png
[20191017151410756.gif]: /images/20230529/ee7081b501f34dd0bbc85963b86f1379.png
[20191017152947800.png]: /images/20230529/64d31ae8ffba4eb18e1b8371db4f5f84.png
[RxJava2]: https://blog.csdn.net/m0_37796683/article/details/102609400
[https_github.com_FollowExcellence_Rxjava_Retrofit]: https://github.com/FollowExcellence/Rxjava_Retrofit