RxJava学习(10):复用与join

前言

根据Froussios英文版的学习笔记。

开发的时候，发现流的转换等部分操作一模一样，此时该如何复用呢？如果发现流的订阅部分操作一模一样，此时又该如何呢？

一、compose与lift

compose转换一个observable为另一个observable，但是lift则是对Subscriber进行更改。

public <R> Observable<R> compose(Observable.Transformer<? super T,? extends R> transformer)

 public final <R> Observable<R> lift(ObservableOperator<? extends R, ? super T> lifter)

 @Test
    public void a() {
        Observable.just(1).compose(transform()).subscribe(System.out::println);
        Observable.just(2).lift(operator()).subscribe(System.out::println);
    }


    @NonNull
    private ObservableTransformer<Integer, String> transform() {
        return upstream -> upstream.map(integer -> "我是第" + integer + "名");
    }


    @NonNull
    private ObservableOperator<String, Integer> operator() {
        return observer -> new Observer<Integer>(){

            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                observer.onNext("我是第" + integer + "名");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };
    }

schedulers

1. immediate ：当前线程立刻开始
2. trampoline：当前线程按照队列顺序执行
3. newThread ：每一个work新建一个线程
4. computation：CPU工作
5. io：IO操作

join

join和zip的区别

zip按顺序结合两个或多个Observables发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。
join则是在任何时候，只要在另一个Observable发射的数据定义的时间窗口内，这个Observable发射了一条数据，就结合两个Observable发射的数据。

zip join