Single

顾名思义，Single 只会接收一个参数，而 SingleObserver 只会调用 onError() 或者 onSuccess()。

Single.just(new Random().nextInt())
                .subscribe(new SingleObserver<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onSuccess(@NonNull Integer integer) {
                        mRxOperatorsText.append("single : onSuccess : "+integer+"\n");
                        Log.e(TAG, "single : onSuccess : "+integer+"\n" );
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        mRxOperatorsText.append("single : onError : "+e.getMessage()+"\n");
                        Log.e(TAG, "single : onError : "+e.getMessage()+"\n");
                    }
                });

输出：

distinct

去重操作符，简单的作用就是去重。

Observable.just(1, 1, 1, 2, 2, 3, 4, 5)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("distinct : " + integer + "\n");
                        Log.e(TAG, "distinct : " + integer + "\n");
                    }
                });

输出：

很明显，发射器发送的事件，在接收的时候被去重了。

debounce

去除发送频率过快的项，看起来好像没啥用处，但你信我，后面绝对有地方很有用武之地。

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                // send events with simulated time wait
                emitter.onNext(1); // skip
                Thread.sleep(400);
                emitter.onNext(2); // deliver
                Thread.sleep(505);
                emitter.onNext(3); // skip
                Thread.sleep(100);
                emitter.onNext(4); // deliver
                Thread.sleep(605);
                emitter.onNext(5); // deliver
                Thread.sleep(510);
                emitter.onComplete();
            }
        }).debounce(500, TimeUnit.MILLISECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("debounce :" + integer + "\n");
                        Log.e(TAG,"debounce :" + integer + "\n");
                    }
                });

输出：

代码很清晰，去除发送间隔时间小于 500 毫秒的发射事件，所以 1 和 3 被去掉了。

defer

简单地时候就是每次订阅都会创建一个新的 Observable，并且如果没有被订阅，就不会产生新的 Observable。

Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> call() throws Exception {
                return Observable.just(1, 2, 3);
            }
        });


        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Integer integer) {
                mRxOperatorsText.append("defer : " + integer + "\n");
                Log.e(TAG, "defer : " + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                mRxOperatorsText.append("defer : onError : " + e.getMessage() + "\n");
                Log.e(TAG, "defer : onError : " + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                mRxOperatorsText.append("defer : onComplete\n");
                Log.e(TAG, "defer : onComplete\n");
            }
        });

输出：

last

Observable.just(1, 2, 3)
                .last(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("last : " + integer + "\n");
                        Log.e(TAG, "last : " + integer + "\n");
                    }
                });

输出：

merge

merge 顾名思义，熟悉版本控制工具的你一定不会不知道 merge 命令，而在 Rx 操作符中，merge 的作用是把多个 Observable 结合起来，接受可变参数，也支持迭代器集合。注意它和 concat 的区别在于，不用等到 发射器 A 发送完所有的事件再进行发射器 B 的发送。

Observable.merge(Observable.just(1, 2), Observable.just(3, 4, 5))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        mRxOperatorsText.append("merge :" + integer + "\n");
                        Log.e(TAG, "accept: merge :" + integer + "\n" );
                    }
                });

输出：

reduce

reduce 操作符每次用一个方法处理一个值，可以有一个 seed 作为初始值。

Observable.just(1, 2, 3)
               .reduce(new BiFunction<Integer, Integer, Integer>() {
                   @Override
                   public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                       return integer + integer2;
                   }
               }).subscribe(new Consumer<Integer>() {
           @Override
           public void accept(@NonNull Integer integer) throws Exception {
               mRxOperatorsText.append("reduce : " + integer + "\n");
               Log.e(TAG, "accept: reduce : " + integer + "\n");
           }
       });

输出：

可以看到，代码中，我们中间采用 reduce ，支持一个 function 为两数值相加，所以应该最后的值是：1 + 2 = 3 + 3 = 6 ， 而Log 日志完美解决了我们的问题。

scan

scan 操作符作用和上面的 reduce 一致，唯一区别是 reduce 是个只追求结果的坏人，而 scan 会始终如一地把每一个步骤都输出。

Observable.just(1, 2, 3)
                .scan(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                mRxOperatorsText.append("scan " + integer + "\n");
                Log.e(TAG, "accept: scan " + integer + "\n");
            }
        });

输出：

看日志，没毛病。

window

按照实际划分窗口，将数据发送给不同的 Observable

mRxOperatorsText.append("window\n");
       Log.e(TAG, "window\n");
       Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次
               .take(15) // 最多接收15个
               .window(3, TimeUnit.SECONDS)
               .subscribeOn(Schedulers.io())
               .observeOn(AndroidSchedulers.mainThread())
               .subscribe(new Consumer<Observable<Long>>() {
                   @Override
                   public void accept(@NonNull Observable<Long> longObservable) throws Exception {
                       mRxOperatorsText.append("Sub Divide begin...\n");
                       Log.e(TAG, "Sub Divide begin...\n");
                       longObservable.subscribeOn(Schedulers.io())
                               .observeOn(AndroidSchedulers.mainThread())
                               .subscribe(new Consumer<Long>() {
                                   @Override
                                   public void accept(@NonNull Long aLong) throws Exception {
                                       mRxOperatorsText.append("Next:" + aLong + "\n");
                                       Log.e(TAG, "Next:" + aLong + "\n");
                                   }
                               });
                   }
               });

输出：

简单的网络请求

想必大家都知道，很多时候我们在使用 RxJava 的时候总是和 Retrofit 进行结合使用，而为了方便演示，这里我们就暂且采用 OkHttp3 进行演示，配合 map，doOnNext ，线程切换进行简单的网络请求：

• 1）通过 Observable.create() 方法，调用 OkHttp 网络请求；
• 2）通过 map 操作符集合 gson，将 Response 转换为 bean 类；
• 3）通过 doOnNext() 方法，解析 bean 中的数据，并进行数据库存储等操作；
• 4）调度线程，在子线程中进行耗时操作任务，在主线程中更新 UI ；
• 5）通过 subscribe()，根据请求成功或者失败来更新 UI 。

Observable.create(new ObservableOnSubscribe<Response>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
                Builder builder = new Builder()
                        .url("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
                        .get();
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                e.onNext(response);
            }
        }).map(new Function<Response, MobileAddress>() {
                    @Override
                    public MobileAddress apply(@NonNull Response response) throws Exception {

                        Log.e(TAG, "map 线程:" + Thread.currentThread().getName() + "\n");
                        if (response.isSuccessful()) {
                            ResponseBody body = response.body();
                            if (body != null) {
                                Log.e(TAG, "map:转换前:" + response.body());
                                return new Gson().fromJson(body.string(), MobileAddress.class);
                            }
                        }
                        return null;
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<MobileAddress>() {
                    @Override
                    public void accept(@NonNull MobileAddress s) throws Exception {
                        Log.e(TAG, "doOnNext 线程:" + Thread.currentThread().getName() + "\n");
                        mRxOperatorsText.append("\ndoOnNext 线程:" + Thread.currentThread().getName() + "\n");
                        Log.e(TAG, "doOnNext: 保存成功：" + s.toString() + "\n");
                        mRxOperatorsText.append("doOnNext: 保存成功：" + s.toString() + "\n");

                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<MobileAddress>() {
                    @Override
                    public void accept(@NonNull MobileAddress data) throws Exception {
                        Log.e(TAG, "subscribe 线程:" + Thread.currentThread().getName() + "\n");
                        mRxOperatorsText.append("\nsubscribe 线程:" + Thread.currentThread().getName() + "\n");
                        Log.e(TAG, "成功:" + data.toString() + "\n");
                        mRxOperatorsText.append("成功:" + data.toString() + "\n");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.e(TAG, "subscribe 线程:" + Thread.currentThread().getName() + "\n");
                        mRxOperatorsText.append("\nsubscribe 线程:" + Thread.currentThread().getName() + "\n");

                        Log.e(TAG, "失败：" + throwable.getMessage() + "\n");
                        mRxOperatorsText.append("失败：" + throwable.getMessage() + "\n");
                    }
                });

为了方便，我们后面的讲解大部分采用开源的 Rx2AndroidNetworking 来处理，数据来源于天狗网等多个公共API接口。

mRxOperatorsText.append("RxNetworkActivity\n");
        Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
                .build()
                .getObjectObservable(MobileAddress.class)
                .observeOn(AndroidSchedulers.mainThread()) // 为doOnNext() 指定在主线程，否则报错
                .doOnNext(new Consumer<MobileAddress>() {
                    @Override
                    public void accept(@NonNull MobileAddress data) throws Exception {
                        Log.e(TAG, "doOnNext:"+Thread.currentThread().getName()+"\n" );
                        mRxOperatorsText.append("\ndoOnNext:"+Thread.currentThread().getName()+"\n" );
                        Log.e(TAG,"doOnNext:"+data.toString()+"\n");
                        mRxOperatorsText.append("doOnNext:"+data.toString()+"\n");
                    }
                })
                .map(new Function<MobileAddress, ResultBean>() {
                    @Override
                    public ResultBean apply(@NonNull MobileAddress mobileAddress) throws Exception {
                        Log.e(TAG, "\n" );
                        mRxOperatorsText.append("\n");
                        Log.e(TAG, "map:"+Thread.currentThread().getName()+"\n" );
                        mRxOperatorsText.append("map:"+Thread.currentThread().getName()+"\n" );
                        return mobileAddress.getResult();
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<ResultBean>() {
                    @Override
                    public void accept(@NonNull ResultBean data) throws Exception {
                        Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName()+"\n" );
                        mRxOperatorsText.append("\nsubscribe 成功:"+Thread.currentThread().getName()+"\n" );
                        Log.e(TAG, "成功:" + data.toString() + "\n");
                        mRxOperatorsText.append("成功:" + data.toString() + "\n");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.e(TAG, "subscribe 失败:"+Thread.currentThread().getName()+"\n" );
                        mRxOperatorsText.append("\nsubscribe 失败:"+Thread.currentThread().getName()+"\n" );
                        Log.e(TAG, "失败："+ throwable.getMessage()+"\n" );
                        mRxOperatorsText.append("失败："+ throwable.getMessage()+"\n");
                    }
                });

先读取缓存，如果缓存没数据再通过网络请求获取数据后更新UI

想必在实际应用中，很多时候（对数据操作不敏感时）都需要我们先读取缓存的数据，如果缓存没有数据，再通过网络请求获取，随后在主线程更新我们的 UI。

concat 操作符简直就是为我们这种需求量身定做。

concat 可以做到不交错的发射两个甚至多个 Observable 的发射事件，并且只有前一个 Observable 终止( onComplete() ) 后才会定义下一个 Observable。

利用这个特性，我们就可以先读取缓存数据，倘若获取到的缓存数据不是我们想要的，再调用 onComplete() 以执行获取网络数据的 Observable，如果缓存数据能应我们所需，则直接调用 onNext() ，防止过度的网络请求，浪费用户的流量。

Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<FoodList> e) throws Exception {
                Log.e(TAG, "create当前线程:"+Thread.currentThread().getName() );
                FoodList data = CacheManager.getInstance().getFoodListData();

                // 在操作符 concat 中，只有调用 onComplete 之后才会执行下一个 Observable
                if (data != null){ // 如果缓存数据不为空，则直接读取缓存数据，而不读取网络数据
                    isFromNet = false;
                    Log.e(TAG, "\nsubscribe: 读取缓存数据:" );
                    runOnUiThread(new Runnable() {
                        @Override
                        public void run() {
                            mRxOperatorsText.append("\nsubscribe: 读取缓存数据:\n");
                        }
                    });

                    e.onNext(data);
                }else {
                    isFromNet = true;
                    runOnUiThread(new Runnable() {
                        @Override
                        public void run() {
                            mRxOperatorsText.append("\nsubscribe: 读取网络数据:\n");
                        }
                    });
                    Log.e(TAG, "\nsubscribe: 读取网络数据:" );
                    e.onComplete();
                }


            }
        });

        Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
                .addQueryParameter("rows",10+"")
                .build()
                .getObjectObservable(FoodList.class);


        // 两个 Observable 的泛型应当保持一致

        Observable.concat(cache,network)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<FoodList>() {
                    @Override
                    public void accept(@NonNull FoodList tngouBeen) throws Exception {
                        Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName() );
                        if (isFromNet){
                            mRxOperatorsText.append("accept : 网络获取数据设置缓存: \n");
                            Log.e(TAG, "accept : 网络获取数据设置缓存: \n"+tngouBeen.toString() );
                            CacheManager.getInstance().setFoodListData(tngouBeen);
                        }

                        mRxOperatorsText.append("accept: 读取数据成功:" + tngouBeen.toString()+"\n");
                        Log.e(TAG, "accept: 读取数据成功:" + tngouBeen.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.e(TAG, "subscribe 失败:"+Thread.currentThread().getName() );
                        Log.e(TAG, "accept: 读取数据失败："+throwable.getMessage() );
                        mRxOperatorsText.append("accept: 读取数据失败："+throwable.getMessage()+"\n");
                    }
                });

有时候我们的缓存可能还会分为 memory 和 disk ，实际上都差不多，无非是多写点 Observable ，然后通过 concat 合并即可。

多个网络请求依次依赖

想必这种情况也在实际情况中比比皆是，例如用户注册成功后需要自动登录，我们只需要先通过注册接口注册用户信息，注册成功后马上调用登录接口进行自动登录即可。

我们的 flatMap 恰好解决了这种应用场景，flatMap 操作符可以将一个发射数据的 Observable 变换为多个 Observables ，然后将它们发射的数据合并后放到一个单独的 Observable，利用这个特性，我们很轻松地达到了我们的需求。

Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
                .addQueryParameter("rows", 1 + "")
                .build()
                .getObjectObservable(FoodList.class) // 发起获取食品列表的请求，并解析到FootList
                .subscribeOn(Schedulers.io())        // 在io线程进行网络请求
                .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理获取食品列表的请求结果
                .doOnNext(new Consumer<FoodList>() {
                    @Override
                    public void accept(@NonNull FoodList foodList) throws Exception {
                        // 先根据获取食品列表的响应结果做一些操作
                        Log.e(TAG, "accept: doOnNext :" + foodList.toString());
                        mRxOperatorsText.append("accept: doOnNext :" + foodList.toString()+"\n");
                    }
                })
                .observeOn(Schedulers.io()) // 回到 io 线程去处理获取食品详情的请求
                .flatMap(new Function<FoodList, ObservableSource<FoodDetail>>() {
                    @Override
                    public ObservableSource<FoodDetail> apply(@NonNull FoodList foodList) throws Exception {
                        if (foodList != null && foodList.getTngou() != null && foodList.getTngou().size() > 0) {
                            return Rx2AndroidNetworking.post("http://www.tngou.net/api/food/show")
                                    .addBodyParameter("id", foodList.getTngou().get(0).getId() + "")
                                    .build()
                                    .getObjectObservable(FoodDetail.class);
                        }
                        return null;

                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<FoodDetail>() {
                    @Override
                    public void accept(@NonNull FoodDetail foodDetail) throws Exception {
                        Log.e(TAG, "accept: success ：" + foodDetail.toString());
                        mRxOperatorsText.append("accept: success ：" + foodDetail.toString()+"\n");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.e(TAG, "accept: error :" + throwable.getMessage());
                        mRxOperatorsText.append("accept: error :" + throwable.getMessage()+"\n");
                    }
                });

结合多个接口的数据更新 UI

在实际应用中，我们极有可能会在一个页面显示的数据来源于多个接口，这时候我们的 zip 操作符为我们排忧解难。

zip 操作符可以将多个 Observable 的数据结合为一个数据源再发射出去。

Observable<MobileAddress> observable1 = Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
                .build()
                .getObjectObservable(MobileAddress.class);

        Observable<CategoryResult> observable2 = Network.getGankApi()
                .getCategoryData("Android",1,1);

        Observable.zip(observable1, observable2, new BiFunction<MobileAddress, CategoryResult, String>() {
            @Override
            public String apply(@NonNull MobileAddress mobileAddress, @NonNull CategoryResult categoryResult) throws Exception {
                return "合并后的数据为：手机归属地："+mobileAddress.getResult().getMobilearea()+"人名："+categoryResult.results.get(0).who;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        Log.e(TAG, "accept: 成功：" + s+"\n");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.e(TAG, "accept: 失败：" + throwable+"\n");
                    }
                });

间隔任务实现心跳

想必即时通讯等需要轮训的任务在如今的 APP 中已是很常见，而 RxJava 2.x 的 interval 操作符可谓完美地解决了我们的疑惑。

这里就简单的意思一下轮训。

private Disposable mDisposable;
    @Override
    protected void doSomething() {
        mDisposable = Flowable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        Log.e(TAG, "accept: doOnNext : "+aLong );
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        Log.e(TAG, "accept: 设置文本 ："+aLong );
                        mRxOperatorsText.append("accept: 设置文本 ："+aLong +"\n");
                    }
                });
    }

    /**
     * 销毁时停止心跳
     */
    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mDisposable != null){
            mDisposable.dispose();
        }
    }

RxJava的另一个好处在于，我们可以清楚地看到数据是如何在一系列操作符之间进行转换的。

Observable.from(someSource)
    .map(new Func1<Data, Data>() {
      @Override public Data call(Data data) {
        return manipulate(data);
      }
    }).subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Data>() {
      @Override public void call(Data data) {
        doSomething(data);
      }
    });

如何将一组操作符重用于多个数据流中呢？例如，因为希望在工作线程中处理数据，然后在主线程中处理结果，所以我会频繁使用subscribeOn()和observeOn()。如果我能够通过重用的方式，将这种逻辑运用到我所有的数据流中，将是一件多么伟大的事。

糟糕的实现方式

下面这些代码是我在过去几个月里一直都在使用的，正好可以拿来当反面教材。

首先，使用schedulers（调度）创建一个方法：

<T> Observable<T> applySchedulers(Observable<T> observable) {
return observable.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread());
}

然后，封装Observable数据链：

applySchedulers(Observable.from(someSource).map(new Func1<Data, Data>() {
      @Override public Data call(Data data) {
        return manipulate(data);
      }
    })
).subscribe(new Action1<Data>() {
  @Override public void call(Data data) {
    doSomething(data);
  }
});

虽然这样做也能达到目的，但是它看起来不仅丑，还容易让人产生困惑，applySchedulers()到底什么鬼？它不再符合操作符链路式结构，所以，看起来很难理解。然而，我找不到任何办法去格式化这段代码，因此，这并不尴尬。

现在，试想一下，如果在一个数据流中反复使用的话，这个反面教材将会变得要多烂有多烂。（译者注：OMG）

Transformers简介

聪明的同学可能已经意识到了这个问题，但是RxJava早已提供了一种解决方案：Transformer（译者注：有转换器意思），一般情况下可以通过使用操作符Observable.compose()来实现。

Transformer实际上就是一个Func1<Observable<T>, Observable<R>>，换言之就是：可以通过它将一种类型的Observable转换成另一种类型的Observable，和调用一系列的内联操作符是一模一样的。

接下来，我们一起创建一个Transformer与schedulers相结合的方法：

 <T> Transformer<T, T> applySchedulers() {
 return new Transformer<T, T>() {
   @Override
   public Observable<T> call(Observable<T> observable) {
     return observable.subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread());
   }
 };
}

使用Lambda表达式，会变得更加优雅:

<T> Transformer<T, T> applySchedulers() {  
return observable -> observable.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread());
}

更新于2015年3月11日：如果使用 JDK 7或者更早版本进行编译的话，就不得不做出一些额外的改变，为compose()添加泛型。显式声明返回类型，就像这样：

 Observable.from(someSource)
    .map(new Func1<Data, Data>() {
       @Override public Data call(Data data) {
         return manipulate(data);
       }
    })
    .compose(this.<YourType>applySchedulers())
    .subscribe(new Action1<Data>() {
      @Override public void call(Data data) {
        doSomething(data);
      }
    });

重用Transformers

上一个实例中，我在所有的方法调用中都new（译者注：新建）了一个Transformer实例。你可以创建一个实例化版本，节省不必要的实例化对象。毕竟，Transformers关乎着代码重用。

如果，总是将一个具体的类型转换成另一个具体的类型，那么可以很容易的创建一个Transformer实例：

Transformer<String, String> myTransformer = new Transformer<String, String>() {  
   // ...Do your work here...
};

那么scheduler Transformer（调度转换器）应该怎样实现呢？因为，它根本不需要关心类型，所以就无法定义一个泛型实例：

// Doesn't compile; where would T come from?（译者注：编译无法通过；因为根本不知道 T 从何而来。）
Transformer<T, T> myTransformer;  

可能有人会这样定义Transformer<Object, Object>，然而这并没有什么卵用，并且造成的后果就是Observable将失去其类型信息。

为了解决这个问题，我从Collections中得到了一些启发，这个包装类有这样一堆方法，能够创建类型安全并且不可变的空集合（比如，Collections.emptyList()）。由于在其内部使用了一个无泛型实例，所以需要封装在一个添加泛型约束的方法里。

可以像下面这样定义我们的schedulers Transformer（调度转换器）实例：

 final Observable.Transformer schedulersTransformer = new  Observable.Transformer() {
 @Override public Object call(Object observable) {
   return ((Observable)  observable).subscribeOn(Schedulers.newThread())
       .observeOn(AndroidSchedulers.mainThread());
  }
};

 <T> Observable.Transformer<T, T> applySchedulers() {
 return (Observable.Transformer<T, T>) schedulersTransformer;
}

现在我们终于把他做成一个“单例”了（译者注：此单例非彼单例）。

警告：如果不做类型转换检查，可能陷入麻烦。确保Transformer真正与类型无关。否则，即使代码通过了编译，在运行时仍然存在抛出ClassCastException异常的隐患。在当前场景中，我们知道它是安全的，因为schedulers（译者注：调度）并不会与发送出的事件产生任何的互动操作。

** flatMap()操作符怎么样？**

现在你可能会好奇，compose()操作符和flatMap()操作符有何区别。他们最终都会发送出Observable<R>，这就意味着，两者都能够用于操作符的重用？

不同点在于compose()操作符拥有更高层次的抽象概念：它操作于整个数据流中，不仅仅是某一个被发送的事件。具体如下：

1. compose()是唯一一个能够从数据流中得到原始Observable<T>的操作符，所以，那些需要对整个数据流产生作用的操作（比如，subscribeOn()和observeOn()）需要使用compose()来实现。相较而言，如果在flatMap()中使用subscribeOn()或者observeOn()，那么它仅仅对在flatMap()中创建的Observable起作用，而不会对剩下的流产生影响（译者注：深坑，会在后面的系列着重讲解，欢迎关注）。

2. 当创建Observable流的时候，compose()会立即执行，犹如已经提前写好了一个操作符一样，而flatMap()则是在onNext()被调用后执行，onNext()的每一次调用都会触发flatMap()，也就是说，flatMap()转换每一个事件，而compose()转换的是整个数据流。

3. 因为每一次调用onNext()后，都不得不新建一个Observable，所以flatMap()的效率较低。事实上，compose()操作符只在主干数据流上执行操作。

如果想重用一些操作符，还是使用compose()吧，虽然flatMap()的用处很多，但作为重用代码这一点来讲，并不适用。

onErrorReturn：

public final Observable<T> onErrorReturn(Func1<java.lang.Throwable,? extends T> resumeFunction)

默认情况下，Observer订阅的Observable如果发生异常或错误，那么Observable就不再释放item而是调用Observer的onError方法之后终止。现在可以使用onErrorReturn来改变这个行为了。可以看到onErrorReturn接受一个Func1实现，当源Observable正常执行时onErrorReturn就直接把源Observable释放的item释放出去，一旦源Observable发生了错误或异常，替代原先源Observable调用Observer的onError方法，onErrorReturn中那个Func1实现被调用并接受这个错误或异常作为参数，这个Func1实现的返回值将作为onErrorReturn返回的Observable的item释放出去然后调用Observer的onCompleted方法终止。

onErrorResumeNext：

public final Observable<T> onErrorResumeNext(Func1<java.lang.Throwable,? extends Observable<? extends T>> resumeFunction)
public final Observable<T> onErrorResumeNext(Observable<? extends T> resumeSequence)

和上面差不多，都是源Observable正常的时候由这个操作符新建的Observable把源Observable释放的item直接释放释放出去。一旦源Observable遇到错误，这个onErrorResumeNext会把源Observable用一个新的Observable替掉，然后这个新的Observable如果没遇到什么问题就会释放item给Observer。你可以直接将一个Observable实例传入onErrorResumeNext作为这个新的Observable，也可以传给onErrorResumeNext一个Func1实现，这个Func1实现接受源Observable的错误作为参数，返回新的Observable。

onExceptionResumeNext：

public final Observable<T> onExceptionResumeNext(Observable<? extends T> resumeSequence)

和上面的onErrorResumeNext差不多，不过触发onErrorResumeNext的是源Observable发生了错误（Error），而触发onExceptionResumeNext的是源Observable发生了异常（Exception），如果源Observable发生了错误仍然像默认情况一样调用Observer的onError方法。

retry：

public final Observable<T> retry()
public final Observable<T> retry(long count)
public final Observable<T> retry(Func2<java.lang.Integer,java.lang.Throwable,java.lang.Boolean> predicate)

默认retry在trampoline Scheduler上执行。正常情况一下retry新建的Observable直接把源Observable释放的item释放出去。当源Observable遇到错误或异常时，对于那个无参retry()方法，ta会重新订阅源Observable，只要源Observable遇到错误或异常ta都会重新订阅源Observable，没次数限制；对于retry(long count)可以限制重新订阅的次数，超过这个次数限制，直接调用Observer的onError终止。还有个retry接受一个Func2实现作为参数，Func2接受两个参数第一个是到现在为止retry新建的Observable已经重新订阅源Observable的次数，第二参数是导致源Observable要调用Observer的onError的错误或异常实例，Fun2的返回值是个Boolean用来指示retry新建的Observable是否再次重新订阅源Observable，如果不再订阅，直接将把源Observable最后的调用onError通知传给Observer。值得注意的是，retry新建的Observable会把源Observable释放的所有item都释放出去，即使中间某次重新订阅发生失败，失败时的源Observable释放的item在下次成功订阅时也会随着释放出去，比如第一次订阅失败了，在此期间源Observable发出[1,2]，但是第二次成功了源Observable释放了[1,2,3,4,5]，那么这次retry新建的Observable释放的完整序列是[1,2,1,2,3,4,5]。

retryWhen：

public final Observable<T> retryWhen(Func1<? super Observable<? extends java.lang.Throwable>,? extends Observable<?>> notificationHandler)
public final Observable<T> retryWhen(Func1<? super Observable<? extends java.lang.Throwable>,? extends Observable<?>> notificationHandler, Scheduler scheduler)

retryWhen默认在trampoline Scheduler上执行，也可以指定ta的Scheduler。源Observable没出状况时直接镜像ta释放的item。一旦源Observable遇到异常，将把这个异常包装成一个Observable传给retryWhen的参数Func1实现，这个Func1实现返回一个Observable，接下来如果这个Observable正常释放一个item，那么retryWhen新建的Observable重新订阅源Obervable，然后镜像释放源Observable释放的item，如果Func1返回的Observable发生错误或异常，发出调用onError通知，那么就直接把这个调用onError通知传递给Observer然后retry新建的Observable终止。