RxJava 转换操作符详解及lift 源码解析

缺乏、安全感 2022-02-24 18:54 264阅读 0赞

RxJava 提供了对事件序列进行变换的支持，这是它的核心功能之一，也是大多数人说『RxJava 真是太好用了』的最大原因。**所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。**特别是map、flatMap、concatMap都是接受一个函数作为参数(Func1)，非常常用，但很多人不知道它们通过 lift() 实现。

**目录**

1. map

2. flatMap

3. concatMap

4. flatMapIterable

5. SwitchMap

6. scan

7. GroupBy

8. compose

9. lift()源码

--------------------

### **1. map** ###

![20190405095001582.png][]

官方的图（正确理解：一种颜色代表一条数据，一个圈转变为一个正方形），要点：

*  map 转换是一对一的，原来发射了几个数据，转换之后还是几个
 *  map 转换可以改变发射的数据类型

map应用场景非常广泛，这里不再列举，先看个例子吧

//String转Integer
    Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
                    subscriber.onNext("123456");
                    subscriber.onComplete();
                }
            }).map(new Func1<String, Integer>() {
                @Override
                public Integer call(String s) {
                    return Integer.parseInt(s);
                }
            }).subscribe(new Consumer<String>() {
                @Override
                public void accept(@NonNull Integer integer) {
                    
                }
            });

### 2. flatMap ###

`flatMap()` 中的 flat 就是所谓『铺平』。将一个发射数据的Observable变换为多个Observables，然后将它们发射的**数据合并**后放进一个单独的Observable（统一路径）。

**原理**

1.  传入的事件对象装换成一个Observable对象
2.  这是不会直接发送这个Observable, 而是将这个Observable激活让它自己开始发送事件
3.  每一个创建出来的Observable发送的事件，都被汇入同一个Observable，这个Observable负责将这些事件统一交给Subscriber的回调方法。

**理解下图精髓：一种颜色的圆代表源数据（Data<List<Item>>），flat map后，每种颜色分裂为多方块（方块代表Item），分裂后的数据是乱序的**

![20190404225514978.png][]  
**应用场景**

*  循环嵌套（循环套循环）
 *  连续请求两个接口（第一个接口的返回值是第二个接口的请求参数），以前我们会在一个请求完成后，在onResponse中获取结果再请求另一个接口。这种接口嵌套，代码看起来是非常丑陋的

public class Student {
        public String name;
        public int id;
        public List<Source> mSources;
    }
    public class Source {
        public int sourceId;//id
        public String name;//课程名
        public int score;//成绩
    }
    //示例
    Flowable.fromIterable(MockData.getAllStudentInfoById(0))
                .flatMap(new Function<Student, Publisher<Source>>() {
                    @Override
                    public Publisher<Source> apply(@NonNull Student student) throws Exception {
                        return Flowable.fromIterable(student.mSources);
                    }
                })   
                .subscribe(new Consumer<Source>() {
                    @Override
                    public void accept(@NonNull Source source) throws Exception {
                        String content = "sourceName:"+source.name +" source score:"+source.score;
                        Log.i(TAG,content);
    
                    }
                });

### **3. concatMap** ###

concatMap()解决了flatMap()的交叉（无序）问题，它能够把发射的值连续在一起，用法及应用场景可以参考flatMap。

图解不再赘述了！！

![20190405101926281.png][]

### **4. flatMapIterable** ###

flatMapIterable()和flatMap()几乎是一样的，不同的是flatMapIterable()它转化的多个Observable是使用Iterable作为源数据的。

Observable.from(communities)
            .flatMapIterable(new Func1<Community, Iterable<House>>() {
                @Override
                public Iterable<House> call(Community community) {
                    return community.houses;
                }
            })
            .subscribe(new Action1<House>() {
    
                @Override
                public void call(House house) {
    
                }
            });

### **5. SwitchMap** ###

switchMap()和flatMap()很像，除了一点：每当源Observable发射一个新的数据项（Observable）时，它将取消订阅并停止监视之前那个数据项产生的Observable，并开始监视当前发射的这一个。

![20190405102351530.png][]

### **6. scan** ###

scan()对一个*序列的数据（list/array）*应用*一个函数（fn1）*，并将这个函数的结果发射出去作为下次运用 fn1 的第一个参数使用。

![20190405102524923.png][]

Observable.just(1, 2, 3, 4, 5)
            .scan(new Func2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer integer, Integer integer2) {
                    return integer + integer2;
                }
            }).subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer integer) {
            System.out.print(integer+“ ”);
        }
    });
    /** 输出结果:
    1 3 6 10 15
    **/

### **7. GroupBy** ###

groupBy()将原始Observable发射的数据按照key来拆分成一些小的Observable，然后这些小Observable分别发射其所包含的的数据，和SQL中的groupBy类似。实际使用中，我们需要提供一个生成key的规则（也就是Func1中的call方法），所有key相同的数据会包含在同一个小的Observable中。另外我们还可以提供一个函数来对这些数据进行转化，有点类似于集成了flatMap。

![20190405103125438.png][]

/**
    假设我现在有一组房源List<House> houses,每套房子都属于某一个小区，现在我们需要根据小区名来对房源进行分类，然后依次将房源信息输出。
    **/
    List<House> houses = new ArrayList<>();
    houses.add(new House("中粮·海景壹号", "中粮海景壹号新出大平层！总价4500W起"));
    houses.add(new House("竹园新村", "满五唯一，黄金地段"));
    houses.add(new House("中粮·海景壹号", "毗邻汤臣一品"));
    houses.add(new House("竹园新村", "顶层户型，两室一厅"));
    houses.add(new House("中粮·海景壹号", "南北通透，豪华五房"));
    Observable<GroupedObservable<String, House>> groupByCommunityNameObservable = Observable.from(houses)
            .groupBy(new Func1<House, String>() {
    
                @Override
                public String call(House house) {
                    return house.communityName;
                }
            });
    //GroupedObservable是一个特殊的Observable，它基于一个分组的key，在这个例子中的key就是小区名
    Observable.concat(groupByCommunityNameObservable)
            .subscribe(new Action1<House>() {
                @Override
                public void call(House house) {
                    System.out.println("小区:"+house.communityName+"; 房源描述:"+house.desc);
                }
            });
    /**
    小区:中粮·海景壹号; 房源描述:中粮海景壹号新出大平层！总价4500W起
    小区:中粮·海景壹号; 房源描述:毗邻汤臣一品
    小区:中粮·海景壹号; 房源描述:南北通透，豪华五房
    小区:竹园新村; 房源描述:满五唯一，黄金地段
    小区:竹园新村; 房源描述:顶层户型，两室一厅
    **/

### 8. compose ###

Observable 还有一个变换方法叫做 compose(Transformer实际上就是一个Func1<Observable<T>, Observable<R>>，可以通过它将一种类型的Observable 转换成另一种类型的Observable )。它和 lift() 的区别在于： lift() 是针对事件项和事件序列的，而 compose() 是针对 Observable 自身进行变换。

observable1
        .lift1()
        .lift2()
        .lift3()
        .lift4()
        .subscribe(subscriber1);
    observable2
        .lift1()
        .lift2()
        .lift3()
        .lift4()
        .subscribe(subscriber2);
    observable3
        .lift1()
        .lift2()
        .lift3()
        .lift4()
        .subscribe(subscriber3);
    observable4
        .lift1()
        .lift2()
        .lift3()
        .lift4()
        .subscribe(subscriber1);
    /**
    这个时候，就应该用 compose() 来解决,可读性、可维护性都提高了
    **/
    public class LiftAllTransformer implements Observable.Transformer<Integer, String> {
        @Override
        public Observable<String> call(Observable<Integer> observable) {
            return observable
                .lift1()
                .lift2()
                .lift3()
                .lift4();
        }
    }
    ...
    Transformer liftAll = new LiftAllTransformer();
    observable1.compose(liftAll).subscribe(subscriber1);
    observable2.compose(liftAll).subscribe(subscriber2);
    observable3.compose(liftAll).subscribe(subscriber3);
    observable4.compose(liftAll).subscribe(subscriber4);

### 9. lift()源码 ###

lift()方法很重要，**针对事件序列的处理和再发送**，很多类似map() flatMap()等都是通过它实现的！

讲述 lift() 的原理只是为了让你更好地了解 RxJava ，从而可以更好地使用它。RxJava 都不建议开发者自定义 `Operator` 来直接使用 lift()，而是建议尽量使用已有的 lift() 包装方法（如 `map、flatMap` 等）进行组合来实现需求，因为直接使用 lift() 非常容易发生一些难以发现的错误。

@Test
    public void go_lift() {
        Observable.just(1).lift(new Observable.Operator<String, Integer>() {
            @Override
            public Subscriber<? super Integer> call(final Subscriber<? super String> subscriber) {
                // 将事件序列中的 Integer 对象转换为 String 对象
                return new Subscriber<Integer>() {
                    @Override
                    public void onNext(Integer integer) {
                        subscriber.onNext("x" + integer);
                    }
                    @Override
                    public void onCompleted() {
                        subscriber.onCompleted();
                    }
                    @Override
                    public void onError(Throwable e) {
                        subscriber.onError(e);
                    }
                };
            }
        }).subscribe(x -> System.out.println(x));
    }

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 进行订阅

![20190405104809679.gif][]

两次和多次的 `lift()` 同理

![20190405104858907.png][]

lift()过程有点**像一种代理机制，通过事件拦截和处理实现事件序列的变换，它是如何实现的呢？**

这就是classA.this的奥秘，看下面代码吧，体会到 lift.action.call("world"); 神奇吧！

//使用 ObservableX.this时，它在内部成员变量中访问时，访问的不是当前作用域的地址
    @Test
    public void go_thisObj() {
        ObservableX observableX = new ObservableX(str -> System.out.println("hello"));
        observableX.display();
        ObservableX lift = observableX.lift();
        lift.display();
        lift.action.call("world");//此次是关键点（代理拦截）
    }
    public class ObservableX {
        Action1<String> action;
    
        public ObservableX(Action1 action) {
            this.action = action;
        }
    
        public void display() {
            System.out.println(ObservableX.this.action);
            System.out.println(action);
        }
    
        public ObservableX lift() {
            return new ObservableX(new Action1<String>() {
                @Override
                public void call(String s) {
                    System.out.println(ObservableX.this.action);
                }
            });
        }
    }
    /**
    com.example.rx.ReactiveX_test$$Lambda$1/863831416@39fb3ab6
    com.example.rx.ReactiveX_test$$Lambda$1/863831416@39fb3ab6
    com.example.rx.ReactiveX_test$ObservableX$1@6276ae34
    com.example.rx.ReactiveX_test$ObservableX$1@6276ae34
    com.example.rx.ReactiveX_test$$Lambda$1/863831416@39fb3ab6
    **/

下面开始看源码吧

/**伪代码
     Observable 执行了 lift(Operator) 方法之后，会返回一个新的 Observable，这个新的 Observable 会像一个代理一样，负责接收原始的 Observable 发出的事件，并在处理后发送给 Subscriber
    **/
    public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
        return Observable.create(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber subscriber) {
                Subscriber newSubscriber = operator.call(subscriber);
                newSubscriber.onStart();
                onSubscribe.call(newSubscriber);
            }
        });
    }
    //源码
    public class Observable<T> {
        public final <R> Observable<R> lift(final Observable.Operator<? extends R, ? super T> operator) {
            return new Observable(new Observable.OnSubscribe<R>() {
                public void call(Subscriber<? super R> o) {
                    try {
                        Subscriber st = (Subscriber)Observable.hook.onLift(operator).call(o);
    
                        try {
                            st.onStart();
                            //这个非常抽象(onSubscribe所指代的对象不同)
                            Observable.this.onSubscribe.call(st);
                        } catch (Throwable var4) {
                            Exceptions.throwIfFatal(var4);
                            st.onError(var4);
                        }
                    } catch (Throwable var5) {
                        Exceptions.throwIfFatal(var5);
                        o.onError(var5);
                    }
    
                }
            });
        }
    }

Map一般用于对原始的参数进行加工处理，concatMap是有序的，flatMap是无序的，他们底层都是通过lift 实现。![89576152][]

[20190405095001582.png]: /images/20220224/0218e8df38564865a7a1d998598ac552.png
[20190404225514978.png]: /images/20220224/095defc6758b4c17aaf52e66e0a64b7d.png
[20190405101926281.png]: /images/20220224/78c5d4e1aa3a4ecf9b087de5a939bdc4.png
[20190405102351530.png]: /images/20220224/52706a5a3e0145d28383162c5737a268.png
[20190405102524923.png]: /images/20220224/62dfb1dd7cce4edab29545008f13f0ae.png
[20190405103125438.png]: /images/20220224/232f8b81e56f4307b8733f6663156240.png
[20190405104809679.gif]: /images/20220224/73ba0533fcb042fc8423dbe22124f090.png
[20190405104858907.png]: /images/20220224/e6188c85ef85492e8106dbe0068ea7cc.png
[89576152]: /images/20220224/9ea24b69909c44cd8fdc8112760c9d49.png