RxJava使用（四）变换

小灰灰 2021-09-16 01:52 399阅读 0赞

[https://blog.csdn.net/tangxl2008008/article/details/51334438][https_blog.csdn.net_tangxl2008008_article_details_51334438]

RxJava 提供了对事件序列进行变换的支持；**所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。**

**       不仅可以针对事件对象，还可以针对整个事件队列。**

**       **变换部分主要来自[《给Android 开发者的 RxJava 详解》][Android _ RxJava]

## 1.   **事件对象变换-map()** ##

`  map()`: 事件对象的直接变换；它是 RxJava 最常用的变换；可以将Observable深入的对象1转换为对象2发送给Subscriber。

基本用法如下：

<table> 
 <tbody> 
 <tr> 
 <td>Observable.just(R.drawable.t) &nbsp; &nbsp; &nbsp; &nbsp; .map(new Func1&lt;Integer, Drawable&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public Drawable call(Integer integer) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return getResources().getDrawable(integer); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; }) &nbsp; &nbsp; &nbsp; &nbsp; .subscribe(new Action1&lt;Drawable&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void call(Drawable drawable) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; showImg(drawable); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; });</td> 
 </tr> 
 </tbody> 
</table>

## **2.    事件序列变换-flatMap()** ##

**flatMap() 也和 map() 相同，也是把传入的参数转化之后返回另一个对象；和 map() 不同的是， flatMap() 中返回的是个 Observable 对象，并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。**

**         flatMap() 的原理：**

**       1. 使用传入的事件对象创建一个 Observable 对象；**

**       2. 并不发送这个 Observable, 而是将它激活，于是它开始发送事件；**

**       3. 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。**

**       这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。**

扩展：由于可以在嵌套的Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。

flatmap()基本用法如下，定义了一个Book类型，包含名称和章节列表；通过flatmap()打印出全部的章节列表。

Ø  Book.java

<table> 
 <tbody> 
 <tr> 
 <td>public class Book { &nbsp; &nbsp; public String name; &nbsp; &nbsp; public List&lt;String&gt; chapterList = new ArrayList&lt;String&gt;(); &nbsp; &nbsp; public void addChapter(String chapter) { &nbsp; &nbsp; &nbsp; &nbsp; chapterList.add(chapter); &nbsp; &nbsp; } }</td> 
 </tr> 
 </tbody> 
</table>

Ø  flagmap()使用

<table> 
 <tbody> 
 <tr> 
 <td>Book[] books = getBookList(5); Observable.from(books) &nbsp; &nbsp; &nbsp; &nbsp; .flatMap(new Func1&lt;Book, Observable&lt;String&gt;&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public Observable&lt;String&gt; call(Book book) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return Observable.from(book.chapterList); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; }) &nbsp; &nbsp; &nbsp; &nbsp; .subscribe(new Action1&lt;String&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void call(String s) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Log.v(TAG, s); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; });</td> 
 </tr> 
 </tbody> 
</table>

### 3.     变换的原理：lift() ###

所有的变换功能可能有所不同，实质上都是针对事件序列的处理和再发送。而在 RxJava 的内部，它们是基于同一个基础的变换方法： lift(Operator)。

lift()实现的源码：

<table> 
 <tbody> 
 <tr> 
 <td>/** &nbsp;* Lifts a function to the current Observable and returns a new Observable that when subscribed to will pass &nbsp;* the values of the current Observable through the Operator function. &nbsp;* &lt;p&gt; &nbsp;* In other words, this allows chaining Observers together on an Observable for acting on the values within &nbsp;* the Observable. &nbsp;* &lt;p&gt; {@code &nbsp;* observable.map(...).filter(...).take(5).lift(new OperatorA()).lift(new OperatorB(...)).subscribe() &nbsp;* } &nbsp;* &lt;p&gt; &nbsp;* If the operator you are creating is designed to act on the individual items emitted by a source &nbsp;* Observable, use {@code lift}. If your operator is designed to transform the source Observable as a whole &nbsp;* (for instance, by applying a particular set of existing RxJava operators to it) use {@link #compose}. &nbsp;* &lt;dl&gt; &nbsp;* &nbsp;&lt;dt&gt;&lt;b&gt;Scheduler:&lt;/b&gt;&lt;/dt&gt; &nbsp;* &nbsp;&lt;dd&gt;{@code lift} does not operate by default on a particular {@link Scheduler}.&lt;/dd&gt; &nbsp;* &lt;/dl&gt; &nbsp;*&nbsp; &nbsp;* @param operator the Operator that implements the Observable-operating function to be applied to the source &nbsp;* &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Observable &nbsp;* @return an Observable that is the result of applying the lifted Operator to the source Observable &nbsp;* @see &lt;a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators"&gt;RxJava wiki: Implementing Your Own Operators&lt;/a&gt; &nbsp;*/ &nbsp; public final &lt;R&gt; Observable&lt;R&gt; lift(final Operator&lt;? extends R, ? super T&gt; operator) { &nbsp; &nbsp; return new Observable&lt;R&gt;(new OnSubscribe&lt;R&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; public void call(Subscriber&lt;? super R&gt; o) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Subscriber&lt;? super T&gt; st = hook.onLift(operator).call(o); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // new Subscriber created and being subscribed with so 'onStart' it &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; st.onStart(); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; onSubscribe.call(st); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (Throwable e) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // localized capture of errors rather than it skipping all operators&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // and ending up in the try/catch of the subscribe method which then &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // prevents onErrorResumeNext and other similar approaches to error handling &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Exceptions.throwIfFatal(e); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; st.onError(e); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (Throwable e) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Exceptions.throwIfFatal(e); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // if the lift function failed all we can do is pass the error to the final Subscriber &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // as we don't have the operator available to us &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; o.onError(e); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; }); &nbsp; }</td> 
 </tr> 
 </tbody> 
</table>

通过lift()源码可以发现，生成了一个新的Observable（new）；并且在新的Observable（new）的OnSubscribe（new）的call()回调方法中，创建了一个新的Subscriber（new），该新的Subscriber（最终目标）使用了最终调用的Subscriber的代理；再使用原来的Observable（old）的onSubscribe（old）来调用新的Subscriber（new）。

![Center][]

图示lift()变换过程-（图片来自[《给Android 开发者的 RxJava 详解》][Android _ RxJava]）

`lift()` 实现过程

1)  lift() 创建一个新Observable ，此时加上之前的原始 Observable，已经有两个 Observable 了；

2) 而同样地，新 Observable 里的新 OnSubscribe 加上之前的原始 Observable 中的原始 OnSubscribe，也就有了两个 OnSubscribe；

3)  当用户调用经过 lift() 后的 Observable 的 subscribe() 的时候，使用的是 lift() 所返回的新的 Observable ，于是它所触发的 onSubscribe.call(subscriber)，也是用的新 Observable 中的新OnSubscribe，即在 lift() 中生成的那个 OnSubscribe；

4) 而这个新 OnSubscribe 的 call() 方法中的 onSubscribe ，就是指的原始 Observable 中的原始 OnSubscribe ，在这个 call() 方法里，新 OnSubscribe 利用 operator.call(subscriber) 生成了一个新的 Subscriber（Operator 就是在这里，通过自己的call() 方法将新 Subscriber 和原始 Subscriber 进行关联，并插入自己的『变换』代码以实现变换），然后利用这个新 Subscriber 向原始 Observable 进行订阅。

lift() 过程，有点像一种代理机制，通过事件拦截和处理实现事件序列的变换。

在 Observable 执行了lift(Operator) 方法之后，会返回一个新的 Observable，这个新的 Observable 会像一个代理一样，负责接收原始的 Observable 发出的事件，并在处理后发送给 Subscriber。

4.   自定义Observable.Operator，直接调用lift()

下面实例实现的功能跟前面**flatMap()**实现的功能一致：

<table> 
 <tbody> 
 <tr> 
 <td>Observable.from(getBookList(4)) &nbsp; &nbsp; &nbsp; &nbsp; .lift(new Observable.Operator&lt;String, Book&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public Subscriber&lt;? super Book&gt; call(final Subscriber&lt;? super String&gt; subscriber) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return new Subscriber&lt;Book&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void onCompleted() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; subscriber.onCompleted(); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void onError(Throwable e) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; subscriber.onError(e); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void onNext(Book book) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; for(String chapter : book.chapterList) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; subscriber.onNext(chapter); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; }) &nbsp; &nbsp; &nbsp; &nbsp; .subscribe(new Action1&lt;String&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void call(String s) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Log.v(TAG, s); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; });</td> 
 </tr> 
 </tbody> 
</table>

5.   compose(Transformer)变换

compose(Transformer)变换，针对Observable 自身进行变换。假设在程序中有多个 Observable ，并且他们都需要应用一组相同的 lift() 变换，可以使用compose方式处理，而不是每个都调用一次相同的变换过程

实例代码，如需要将Book转换得到对应的全部章节，两个Observable对章节数据有不同的处理方式：

<table> 
 <tbody> 
 <tr> 
 <td>final Observable.Transformer bookTtransformer = new Observable.Transformer&lt;Book, String&gt;() { &nbsp; &nbsp; @Override &nbsp; &nbsp; public Observable&lt;String&gt; call(Observable&lt;Book&gt; bookObservable) { &nbsp; &nbsp; &nbsp; &nbsp; return bookObservable.flatMap(new Func1&lt;Book, Observable&lt;String&gt;&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public Observable&lt;String&gt; call(Book book) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return Observable.from(book.chapterList); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; &nbsp; }); &nbsp; &nbsp; } }; &nbsp;&nbsp; Observable.from(getBookList(4)) &nbsp; &nbsp; &nbsp; .compose(bookTtransformer) &nbsp; &nbsp; &nbsp; .subscribe(new Action1&lt;String&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void call(String s) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; //处理方式1 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Log.v(TAG, s); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; }); &nbsp;&nbsp; Observable.from(getBookList(2)) &nbsp; &nbsp; &nbsp; .compose(bookTtransformer) &nbsp; &nbsp; &nbsp; .subscribe(new Action1&lt;String&gt;() { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public void call(String s) { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; //处理方式2 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Log.v(TAG, s); &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } &nbsp; &nbsp; &nbsp; });</td> 
 </tr> 
 </tbody> 
</table>

[https_blog.csdn.net_tangxl2008008_article_details_51334438]: https://blog.csdn.net/tangxl2008008/article/details/51334438
[Android _ RxJava]: http://gank.io/post/560e15be2dca930e00da1083
[Center]: /images/20210527/41c75910d5374f1d89ac55b75dd079db.png