一. 前言

我尽力做到让这篇文章成为全网最通俗易懂，最细致全面的RxJava使用教程。

在这篇文章中，你将会看到全网最细致的map与flatMap讲解
。。。

前面的文章中，我已经把RxJava是什么以及它的好处，讲的非常清楚了。大家现在是不是热情高涨，迫不及待想要开始实操了？又或者平时虽然其实经常用RxJava，但并不知道它的“标准”使用
所谓“标准”使用，就是一个模板，涵盖RxJava的绝大部分使用方向

这篇文章，就是来解决这些问题的。

学完这篇文章，你将对RxJava的功能有更加直观的理解，对RxJava的使用方向熟记在心。话不多说，我们直接开始吧！

本篇文章有数万字，可能需要花费你1h的时间来进行研究，不过，我向你保证，如果你之前没有接触过RxJava的使用，又或者对它的使用有些模糊，那么，这1h是绝对值得你花费的。

二. 一个标准使用案例

所谓标准使用案例，就是一个涵盖了这个框架基本功能的代码。这个代码会非常简洁，结构非常清晰，让你可以时不时地拿来翻阅，看完后就觉得思路清晰。

需要说明的是，这里只讲了RxJava2的使用案例，并不会涉及RxJava1和RxJava3。原因是RxJava1比较久远，同时大家对RxJava2比RxJava3更熟悉。

标准案例代码

对于RxJava2，其实是有两套使用方式，一是Observable/Observer，二是Flowable/Subscriber。两者的使用方式总体上一样，没有什么大的差别。我们先以Observable/Observer这一对儿为基础讲解，等理透了这一对的思路后，你自然也就掌握了Flowable/Subscriber的基本使用（两者使用起来很像，所以在本篇文章中，以Observable/Observer为重点来进行讲解）。

Observable.create(new ObservableOnSubscribe<Integer>() {


  @Override






  public void subscribe(ObservableEmitter<Integer> e) throws Exception {


    e.onNext(1);


    e.onNext(10);

    e.onComplete();

  }

}).subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .delay(5, TimeUnit.SECONDS)
    .map(new Function<Integer, Integer>() {
      @Override
      public Integer apply(Integer integer) throws Exception {
        return integer + 1;
      }
    })
    .subscribe(new Observer<Integer>(){
      Disposable mDisposable;
      @Override
      public void onSubscribe(Disposable d) {
        mDisposable = d;
      }



      @Override
      public void onNext(Integer value) {
        if (value == 10) {
          mDisposable.dispose();
        }
      }


      @Override
      public void onError(Throwable e) {
      }

      @Override
      public void onComplete() {
      }
    });

这就是一个非常非常标准的RxJava的使用代码，心急的人可能看到有这么一长串，心里已经烦躁了，我知道你很急，但你先别急，请你认真的一行一行看完，看完后，记不住，不理解都没有关系，因为接下来我将会为你划分结构，你一看就会恍然大悟。但前提是你必须一行一行看完案例代码，这样才能达到最佳效果。

看到这里，你是否感觉清晰了很多？没错，RxJava的结构就是这么清晰。如果你觉得还不够简洁明了的话，可以反复看看这张图，然后来看我下面的解释。

关于RxJava的使用，我划分成了以下几个部分。

• 创建被观察者并定义发送事件的方式
• 创建观察者
• 数据转换
• 线程切换
• 解除订阅
• 其他功能

这些点，都与上图一一对应。这样划分，逻辑是非常清晰的。即先知道RxJava两大角色怎么使用，再知道中途数据流动过程中的一些处理方式，最后再知道解除的方式，齐活儿！

下面，我就一个一个地来介绍～

使用点1：创建被观察者并定义发送事件的方式

我们已经知道，RxJava有个非常著名的地方，就是观察者模式。观察者模式的两大角色，就是被观察者和观察者。在RxJava这里，也不例外。我们需要知道RxJava创建被观察者的几种方式。虽然方式有很多，但本质都是一样的，就是创建Observable。所以大家也不必害怕，同时，RxJava在创建被观察者对象时，往往也定义了发送事件的方式，所以这两个可以组合为一个点讲解。

下面我将从简单的方式开始，避免把大家弄迷糊。

通用的创建被观察者的方式

非常简单，直接看代码

Observable.create(new ObservableOnSubscribe<T>() {
  @Override






  public void subscribe(ObservableEmitter<T> e) throws Exception {
      // 定义发送事件的方式
  }


})

这种方式不仅简单，而且也是非常常用的一种方式。

在上面的代码中，我们通过调用Observable的create方法来创建被观察者。在create方法里面，需要传入一个ObservableOnSubscribe<T>类型的参数，其中<T>表示要发送的数据的类型，比如Integer等。并且也实现了subscribe方法，此方法的作用就是定义事件发送的方式。要想发送事件，就调用e.onNext()方法即可。比如：

Observable.create(new ObservableOnSubscribe<Integer>() {


  @Override






  public void subscribe(ObservableEmitter<Integer> e) throws Exception {


    e.onNext(1);


    e.onNext(10);

    e.onComplete();

  }

})

这样之后，我们就创建了一个被观察者对象。这种方式也是非常标准的一种方式，结构清晰，也方便自定义一些实现。接下来，我将说一些不是那么主流的创建方式，但在实际的项目中，却时有用到。

其他的创建被观察者的方式

just

比如

Observable.just(1, 2, 3)

可以依次发射参数，发送完后会自动调用onComplete()方法终止。
在这里就是会自动发射整数1、2、3，并在发射完成后自动调用onComplete()方法。

其实这种方式，等价于

Observable.create(new ObservableOnSubscribe<Integer>() {


  @Override






  public void subscribe(ObservableEmitter<Integer> e) throws Exception {


    e.onNext(1);


    e.onNext(2);
    e.onNext(3);
    e.onComplete();
  }

})

只不过它写起来确实要简单一些，所有有时可以直接使用 just。

使用 just 时，需要注意一点，即它并不是随意塞参数的。最多只能接受10个参数。结论来自源码：

所以，它只适用于参数较少的情况。如果参数确实比较多，那就可以使用下一种方式

fromArray

比如

Observable.fromArray(new int[]{1, 2, 3});

这样就没有像 just 那样严格的大小限制了，它会依次发送数组中的每个元素。

其他的

除了 just 和 fromArray，其实还有很多其他的，在这里就不一一介绍了，目的都是一样的，就是创建被观察者，同时定义发送事件的方式。至于具体有哪些，你可以点进Observable，看它里面都有哪些静态方法，这些基本就都是了

使用点2：创建观察者的方式

通用的创建观察者的方式

其实在上面的代码示例中也已经给出了，

new Observer<Integer>(){
  @Override






  public void onSubscribe(Disposable d) {
  }

  @Override
  public void onNext(Integer value) {
  }



  @Override
  public void onError(Throwable e) {
  }



  @Override
  public void onComplete() {
  }
}

Observer其实是一个接口，我们通过new Observer来创建它的实现类，一共需要实现四个方法，onSubscribe、onNext、onError和onComplete。

（1）onSubscribe是发生订阅的回调，即调用subscribe方法，Observer订阅Observable时，首先会调用onSubscribe方法。我们可以在这里做一些初始化相关的工作。

（2）onNext是Observable来调用的。一般是真正的发送事件。比如上面例子中的，发送一个Integer类型数据。Observer的onNext方法可以接收到这个数字，对其进行一些处理。

（3）onError，见名知意，就是发生错误时调用。

（4）onComplete，见名知意，就是事件流结束时调用。

这就是一个非常标准的观察者的创建方式。下面我们来看一点其他的创建方式

其他的创建观察者的方式

当我们在Observable类中搜索subscribe方法时，可以看到它不仅可以接受Observer类型的参数（即上面讲到的标准的创建方式），还可以接受Consumer类型的参数。如图

所以Consumer就是我们要讲到的第二种创建观察者的方式。

new Consumer<Integer>() {
  @Override






  public void accept(Integer integer) throws Exception {

  }


}

可以看到这里只有一个实现方法，即accept。你可以简单的理解为，这个accept就是指代的onSubscribe、onNext、onError。具体是指代哪一个，需要看我们调用subscribe的方式。从上面那张图中我们可以看出subscribe可以接收多个Consumer，也可以接收一个Consumer。accept到底指代的哪一个方法，需要从subscribe方法定义上找答案。

---

当只有一个Consumer参数时，accept代表的是onNext

从参数名可以看出来

当有两个Consumer参数时，第一个accept代表onNext，第二个accept代表onError

当有三个Consumer参数时，第一个代表onNext，第二个代表onError，第三个代表onSubscribe

---

值得一提的是，onComplete方法并无法通过Consumer替代，而是Action类型。

彩蛋：Consumer和Observer是什么关系？

我们这个小节在讲创建观察者的几种方式。首先讲到了创建Observer的实现类，然后讲到了Consumer。说明这两个都可以作为“观察者”，但这两个有什么区别呢？其实这里很有意思，我们还是从源码中找答案。

当我们调用subscribe方法，传入Consumer参数的时候，实际内部会调用

subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe

依据在这里

我们看看这个方法内部做了什么事情

首先进行了各种判空，然后，把Consumer作为构造参数，创建了LambdaObserver对象！然后调用了subscribe方法，把LambdaObserver对象作为参数传进去了！而subscribe方法，就是我们创建Observer的时候调用的方法

在这里我们可以得出结论：Consumer最后还是会被封装为Observer！

也即，不管是Observer，还是Consumer，最终都会成为Observer，并且调用到subscribe(Observer<? super T> observer)方法完成订阅，只不过Consumer内部作了一层转换，被包成了LambdaObserver（其实LambdaObserver也是一个实现了Observer接口的实现类）

以上，就把RxJava创建观察者和被观察者的方式讲解完成了。下面我们来看在数据流动的过程中，RxJava可以做哪些事情。

使用点3：数据转换的方式

我们在上面的标准案例中，写到了一个数据转换的方式，就是map操作符，所以我们首先来看看这个map操作符的作用

map操作符

我们在java中，也用到过 map，比如 HashMap。我们都知道，它是通过键值对存储，一个 key 对应一个 value。或许我们也可以这样理解，即一个 key 可以转换成一个 value。key 和 value 的类型或许不同，但 value 可以由 key 转换而来，即key可以转换成一个value。

理解了这一点，map操作符也就理解了。它的功能也是完成转换，也是将一个值转换为另外一个值，两个值类型可以不同，但对应着一个转换关系。比如下面的例子

.map(new Function<Integer, Integer>() {
  @Override






  public Integer apply(Integer integer) throws Exception {
    return integer + 1;
  }


})

里面的apply方法，你可以理解为就是定义转换规则的方法。参数类型是Integer，方法也返回了Integer类型。也就是说它将一个Integer类型的数据转换成了一个Integer类型的数据，具体的转换规则是将输入的Integer数据加1。输入的Integer（即方法参数）就类似于 map 中的 key，输出的Integer就类似于 map 中的 value。

map方法里面需要传入一个Function类型的参数，在Function类中，有两个泛型，前一个代表输入的数据类型，后一个代表输出的数据类型。比如

红圈表示Integer类型，蓝圈表示String类型。整个数据的转换，是由Integer类型转为了String类型。

对于map的使用，你是否已经很清楚了呢？下面我们来看一个和map很像的操作符：flatMap

flatMap操作符

关于flatMap，网上的文章众说纷纭，我看了很多，感觉都很懵，所以在这里，我建议大家忘掉之前对flatMap的印象，从0开始跟着我的思路掌握它，你会发现非常的清晰。

本小节内容比较多，但我向你保证，你只需要花10分钟，认真理解透彻举例内容，你就会完全搞懂flatMap的使用。我保证，这是你没有见过的角度，而且这个角度非常通俗易懂。但前提是你需要花时间看完。

我们首先需要拿它和map操作符进行一个对比。对比的方式，就看源码定义吧：

map的源码定义

我们需要注意的点有三个部分，我都用红框的形式标注出来了。

第一是注释，第二是map方法的返回值类型，第三是map方法的Function参数中泛型的定义。让我们来一一解读一下。

首先注释部分，写到参数是一个函数，应用于ObservableSource发出的每个项。返回值是一个Observable对象，它从源ObservableSource中发出项目，并通过指定的函数进行转换。我翻译的比较生硬，但也大概可以看懂。

然后是map的返回值类型，是一个Observable类型

然后是Function的泛型。这里的泛型，除了要求第一个泛型数据，要和上游传下来的数据类型相同外（本文前面的内容有讲解），基本没有明确的限制。

好，接下来，让我们看一下 flatMap 的源码定义

flatMap的源码定义

同样的，我们依然需要关注这三个部分。

同样的，我先来解读一下注释：参数是一个函数，当应用于源ObservableSource发出的项时，返回一个ObservableSource（这里和map就不同了）。返回值是一个Observable对象，它发出将转换函数应用于源ObservableSource发出的每个项的结果，并合并从该转换获得的ObservableSource的结果（这里的合并，你不用刻意去理解，先忽视这个词，不然容易跑偏）

然后是 flatMap 的返回值类型，依然是一个Observable类型

然后是Function的泛型，这里的泛型，除了有和 map 相同的限制外，还多了一个限制，即? extends ObservableSource<? extends R>。即要求返回值，比如是ObservableSource类型的数据。

看到这，你是不是有点懵？懵就对了，但我们要清楚我们懵的点，其实我们懵的点在于，对 map 和 flatMap 的使用区别，不清楚。但我们对 map 和 flatMap 本身的定义，已经很清楚了，就是上文的列举。在这里，我再总结一下：flatMap的Function参数，内部的apply方法，要求返回一个ObservableSource类型的数据，而 map 没有这样的要求，仅此而已，其他的基本完全一样。

为了验证此观点，我们可以写两个demo看下执行效果。首先，来看 flatMap 的例子：

Observable.just("1", "2", "3")

    .flatMap(new Function<String, ObservableSource<String>>() {
      @Override



      public ObservableSource<String> apply(String s) throws Exception {
        return Observable.just(s); // 注意这里的返回值
      }



    }).subscribe(new Observer<String>() {

      @Override

      public void onSubscribe(Disposable d) {

        Log.v("tag", "onSubscribe");

      }



      @Override

      public void onNext(String s) {

        Log.v("tag", "onNext s:" + s);

      }



      @Override

      public void onError(Throwable e) {




      }



      @Override

      public void onComplete() {

        Log.v("tag", "onComplete");

      }

    });

然后，再看个 map 的例子

Observable.just("1", "2", "3")

    .map(new Function<String, String>() {
      @Override



      public String apply(String s) throws Exception {
        return s; // 注意这里的返回值
      }



    }).subscribe(new Observer<String>() {

      @Override

      public void onSubscribe(Disposable d) {

        Log.v("tag", "onSubscribe");

      }



      @Override

      public void onNext(String s) {

        Log.v("tag", "onNext s:" + s);

      }



      @Override

      public void onError(Throwable e) {




      }



      @Override

      public void onComplete() {

        Log.v("tag", "onComplete");

      }

    });

当运行后，你会发现，两者的结果完全一样！都是：

onSubscribe
onNext s:1
onNext s:2
onNext s:3
onComplete

所以，map 和 flatMap 定义上的区别，就是Function要求的返回值不同。那这两个的使用场景，又有什么区别呢？别急，马上就要经历大彻大悟的感觉了！

首先，我们看这样一个使用场景，有一群人在排队，每个人，身上都带着一些水果。在队伍的旁边，有一位大佬，名字叫Observer，Observer需要得到所有人的水果，并逐个显示出来。同时还存在一个检查员的角色，它的作用就是得到每个人身上的水果，并且发送给Observer。至于怎么得到，怎么发送，就是 map 和 flatMap 的区别。我再强调一遍，Observer的要求：得到每一个人的水果，并且逐个显示出来。

用map来实现这样的功能，代码应该这样写：

Observable.fromIterable(people)

    .map(new Function<Man, List<String>>() {
      @Override



      public List<String> apply(Man source) throws Exception {
        return source.fruits;
      }



    })

    .subscribe(new Observer<List<String>>() {
         @Override

         public void onSubscribe(Disposable d) {



         }




         @Override

         public void onNext(List<String> s) {
           for(String str: s) {
            Log.v("tag", "onNext s:" + str);
           }
         }



         @Override
         public void onError(Throwable e) {



         }

         @Override
         public void onComplete() {

         }
       }
    );

people就是人的集合，也就是例子中的队伍，source对象就是每一个人，它的fruits属性，就是这个人身上的水果。从这个例子中，可以看到，map 把每一个人身上的水果都一次性全部拿到，并且整体全部丢给Observer。

如果用 flatMap 来实现这样的功能呢？

Observable.fromIterable(people)

    .flatMap(new Function<Man, ObservableSource<String>>() {
      @Override



      public ObservableSource<String> apply(Man student) throws Exception {
        return Observable.fromIterable(student.fruits);
      }



    })

    .subscribe(new Observer<String>() {
         @Override

         public void onSubscribe(Disposable d) {



         }




         @Override

         public void onNext(String s) {
           Log.v("tag", "onNext s:" + s);
         }

         @Override
         public void onError(Throwable e) {


         }



         @Override
         public void onComplete() {

         }
       }
    );

flatMap 也是一次性把这个人身上的水果全部拿到，但是它和 map 有一个不同，即它把水果拿到之后，会放到一个水果篮子里，这个水果篮子里包括了这个人身上所有的水果，然后，再依次地，一个一个地，递给Observer。

大家请仔细看下上面的两段代码，再继续往后读。

好的，读到这里，我相信你心里已经大体知道两者的区别了，我在这里，再总结一下。接着刚刚的例子讲，map 和 flatMap，就类似这样的区别：

检查员map，检查一个人，就把他身上的所有水果直接丢给Observer，Observer一次性全部接住。

检查员flatMap，检查每个人时，先把他身上的水果，放到一个篮子里，然后再把篮子里的所有水果，依次给Observer，Observer一个一个地接住，等接完了，检查员再检查下一个人。

这两种方式虽然能实现一样的最终效果，但经历的过程不同。不同点在于，有没有把水果，放到一个篮子里的操作，同时将水果丢给Observer时，是一次性全部丢给他，还是一个一个地给他。

怎么样？是不是对两者的功能区别，已经非常清晰了？在这里，其实大家也可以发现一个点，就是 map 和 flatMap 并没有网上说的那么玄幻的差别，其实 flatMap 能实现的作用，map 基本也能实现（比如上面的例子，执行结果是一样的），只是两者侧重点不同。

至此，也可以解释下 flatMap 的注释的含义了，我再贴一下注释代码，省的你往上翻了

这里的所谓合并（merging），就是例子中说到的，把每一个人身上的所有水果，放到一个篮子里的操作，这就是合并。结合上面的例子，我相信你能理解透彻 flatMap 的定义了。

读到这里，我相信你心里已经非常清楚两者功能上的区别了，接下来我们走到最后一步，即两者的实际使用场景，有何不同？或者说，flatMap 一般用在哪里？

我就直接揭晓答案了：map适合于一对一的数据转换，flatMap适合于一对多的数据转换，并且要求Observer将数据一个一个地进行接收！

比如，在上面我举的水果例子中，其实就是个一对多的数据转换，即一个人，转换为多个水果，并且Observer更适合一个一个的接收（谁也不想被人一次性扔过来一堆水果还得全接住吧），这里其实就是比较适合用 flatMap。再比如，要从数据库中，根据 id，取得 id 对应的很多字段，这里相当于是一个一对多的转换，即 id，转换为了 id 对应的很多字段，如果下游再要求最好能一个一个字段给我的话，那这里就非常适合用 flatMap 了

好了，关于 map 和 flatMap 的区别，就讲到这里了。我相信，通过水果的这个例子，你能印象深刻地记住 map 和 flatMap 的区别，同时，也会知道什么场景用 map，什么场景用 flatMap 了。其实除了 map 和 flatMap ，还有其他的数据转换方式，在这里就不一一列举了，最常用的，其实就是 map。

使用点4：线程切换的方式

在这篇文章中，我其实有说到RxJava能够极其方便的实现线程切换，它实现线程切换，主要是两个方法：subscribeOn和ObserveOn。

observeOn 用于指定观察者（Observer）执行的线程，而 subscribeOn 用于指定被观察者（Observable）执行的线程

observeOn是指定一次生效一次，并且只对后面的Observer生效，subscribeOn是只有第一次指定生效，后续指定无效。

至于指定什么线程，RxJava给了两个类：Schedulers和AndroidSchedulers，这两个类可以快捷选择线程，具体有

关于Schedulers：

下面我们来一个一个地解释，

1. Schedulers.single()：只有一个工作者线程来执行任务，如果连续多次调用了Schedulers.single()，那么会确保被指定线程的任务，按顺序一个接一个地执行，并且他们都在同一个线程里面。适用于需要保持任务执行顺序的情况。
2. Schedulers.computation()：这是一个用于计算密集型任务的线程池。它适合用于执行需要大量计算的操作，比如数据处理或复杂的数学运算。
3. Schedulers.io()：这是一个用于 I/O 操作的线程池，适用于执行网络请求、文件读写等耗时的异步任务。
4. Schedulers.trampoline()：这个调度器会按照队列顺序在当前线程上执行任务，适用于需要在当前线程上顺序执行的场景。
5. Schedulers.newThread()：每个任务都在独立的新线程上执行，适用于执行一些比较耗时的异步任务。

关于AndroidSchedulers

我们只需要关注它的AndroidSchedulers.mainThread()即可。是用于 Android 应用的特殊调度器，用于在主线程上执行任务，主要用于 UI 更新和界面处理。

让我们来看一个使用案例吧

Observable.create(new ObservableOnSubscribe<List<String>>() {
      @Override
      public void subscribe(ObservableEmitter<List<String>> emitter) throws Exception {
        Log.v("tag", "subscribe方法，当前线程为: " + Thread.currentThread().getName());
        Thread.sleep(1000);
        List strList = new ArrayList<String>();
        strList.add("1");
        strList.add("2");


        Log.v("tag", "subscribe方法，拿到数据了，下面进行数据发送");
        emitter.onNext(strList);
      }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<List<String>>() {
         @Override
         public void onSubscribe(Disposable d) {

         }


         @Override
         public void onNext(List<String> s) {
           Log.v("tag", "onNext方法，当前线程为: " + Thread.currentThread().getName());
           Log.v("tag", "onNext方法，收到的数据为: " + s.toString());
         }

         @Override
         public void onError(Throwable e) {


         }

         @Override
         public void onComplete() {

         }
       }
    );

执行结果：

看到这里，你也可以发现，使用RxJava，确实可以实现极其方便的线程转换，即一行代码切一个线程，并且还可以保证执行的先后顺序。这给我们带来了极大的方便，尤其是那种先获取网络请求，再更新UI的场景，用RxJava来写，简直太香了！

使用点5：解除订阅的方式

当我们的Observer，不需要再接收数据的时候，我们就可以解除订阅了，解除订阅主要是有两个目的：防止内存泄露，减少不必要的资源消耗。

解除订阅的方式很简单，在开头的标准案例中其实也有，我再拿过来，大家看一下就会了

涉及解除订阅的代码逻辑部分，我都用红框标注出来了。还记得我开头讲过吗？除了Observable/Observer这一对，其实还有Flowable/Subscriber这一对，那么后者是如何解除订阅呢？
且看：

其实这里，使用上很简单，但就是需要我们记住：千万别忘了解除订阅！！！

其实Observable/Observer和Flowable/Subscriber的解除订阅的原理是不同的，这里其实很有意思，等我们讲到原理课的时候，会为大家解释。

使用点6：其他功能

除了以上讲到的RxJava的使用外，其实RxJava还有一些其他的宝藏功能，在这里，我就不一一介绍了，只简单举一些例子，供你参考。

比如它可以实现数据延时发送

它也可以设置超时时间，如果 Observable 在指定的时间内没有发射任何数据项，那么它会抛出一个 TimeoutException，从而能够处理超时情况。

除了这些，其实还有，我就不一一列举了。

三. 总结

我通过差不多两万字的内容，向你详细介绍了RxJava的使用，在文章开头，我先给了你一个标准使用案例，然后以此为线索，依次介绍RxJava各个方向的使用，其中花了很大篇幅讲解关于 map、flatMap 的使用。

如果本篇文章你觉得写的还不错的话，我希望你可以把它当作RxJava的“词典”，每当忘记某一个地方怎么写的时候，就可以拿来翻阅。当然，有问题欢迎在评论区交流讨论。

后面，我们就开启RxJava原理的学习了！