RxJava 操作符
阅读本文前请先了解 RxJava 的基本使用。
参考文档：
Observable 的创建
Observable 对象。可用来遍历集合。
方法列表：
public static
Observable from(Future extends T> future, long timeout, TimeUnit unit) public static
Observable from(Future extends T> future, Scheduler scheduler)
栗子：
// 1. 遍历集合Observableobservable = Observable.from(new String[]{"hello", "hi"});
// 2. 使用 Future 创建 Observable，Future 表示一个异步计算的结果。FutureTaskfutureTask = new FutureTask (new Callable () { @Override public String call() throws Exception { // TODO 执行异步操作并返回数据 return "hihi"; } }); Scheduler.Worker worker = Schedulers.io().createWorker(); worker.schedule(new Action0() { @Override public void call() { futureTask.run(); } }); Observable observable = Observable.from(futureTask);
Observable 对象。
方法列表：
public static
Observable just(T t1, T t2, T t3, T t4, T t5, T t6) public static
Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7) public static
Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8) public static
Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9) public static
Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9, T t10)
栗子：
Observableobservable = Observable.just("hello");// 使用 just() 遍历几个元素Observable observable = Observable.just("hello", "hi", "...");// 使用 from() 方法遍历，效果和 just() 一样。String[] stringArrs = new String[]{"hello", "hi", "..."}; Observable observable = Observable.from(stringArrs);
just() 方法可传入 1~10 个参数，也就说当元素个数小于等于 10 的时候既可以使用 just() 也可以使用 from()，否则只能用 from() 方法。
Observable 对象。
方法列表：
@Beta public static
Observablecreate(SyncOnSubscribe syncOnSubscribe)@Experimental public static
Observablecreate(AsyncOnSubscribe asyncOnSubscribe)
栗子：
Observable.OnSubscribeonSubscribe = new Observable.OnSubscribe() { @Override public void call(Subscriber super String > subscriber) { // onNext() 方法可执行多次 subscribe.onNext("hello"); subscribe.onCompleted(); } }; Observable
此方法不常用，大多数时候都是使用 just( )、form( ) 等方法，如上面那串代码就可以写成：
Observable
Observable 对象，可用来做倒计时等操作。
方法列表：
public static Observable
interval(long interval, TimeUnit unit) public static Observable
interval(long interval, TimeUnit unit, Scheduler scheduler) public static Observable
interval(long initialDelay, long period, TimeUnit unit)
栗子：
// 每隔 1 s 发送一个序列号，序列号从 0 开始，每次累加 1。 Observableobservable = Observable.interval(1, TimeUnit.SECONDS);
Observable 对象，可用来做定时操作。
方法列表：
栗子：
// 定时 3 s Observableobservable = Observable.timer(3, TimeUnit.SECONDS);
Observable 对象。
方法列表：
栗子：
// 依次发射 5、6、7Observableobservable = Observable.range(5, 3);
Observable 对象。
方法列表：
栗子：
// 发出一个 onCompleted() 通知Observable
Observable 对象。
方法列表：
栗子：
// 发出一个 onError() 通知Observable
Observable 对象。
方法列表：
栗子：
Observable
Observable 对象。
方法列表：
栗子：
Observableobservable = Observable.defer(new Func0 >() { @Override public Observable call() { return Observable.just("string"); } });
public final Observable repeat()
public final Observable
栗子：
Observableobservable = Observable.just("string");// 无限重复执行observable.repeat();// 重复执行 5 次observable.repeat(5);
public final Observable repeatWhen(final Func1 super Observable extends Void>, ? extends Observable>> notificationHandler)
栗子：
observable.repeatWhen(new Func1, Observable>>() { @Override public Observable> call(Observable extends Void> observable) { // 重复 3 次, 每次间隔 1 s return observable.zipWith(Observable.range(1, 3), new Func2 () { @Override public Integer call(Void aVoid, Integer integer) { return integer; } }).flatMap(integer -> Observable.timer(1, TimeUnit.SECONDS)); } });
Observable对象的序列出现异常时，不直接发出 onError() 通知，而是重新订阅该 Observable对象，直到重做过程中未出现异常，则会发出 onNext() 和 onCompleted() 通知；若重做过程中也出现异常，则会继续重试，直到达到重试次数上限，超出次数后发出最新的 onError() 通知。
方法列表：
栗子：
Observableobservable = Observable.create(new Observable.OnSubscribe () { @Override public void call(Subscriber super Integer> subscriber) { System.out.println("......."); int a = 1 / 0; subscriber.onNext(a); subscriber.onCompleted(); } });// 无限次的重试observable.retry();// 重试 3 次observable.retry(3);// 使用谓语函数决定是否重试observable.retry(new Func2 () { @Override public Boolean call(Integer integer, Throwable throwable) { // 参数 integer 是订阅的次数; 参数 throwable 是抛出的异常 // 返回值为 true 表示重试, 返回值为 false 表示不重试 return false; } });
public final Observable retryWhen(final Func1 super Observable extends Throwable>, ? extends Observable>> notificationHandler)
栗子：
// 重试 3 次，每次间隔 1 sobservable.retryWhen(new Func1, Observable>>() { @Override public Observable> call(Observable extends Throwable> observable) { return observable.zipWith(Observable.range(1, 3), new Func2 () { @Override public Object call(Throwable throwable, Integer integer) { return integer; } }).flatMap(new Func1
Observable 发射的元素应用于指定的函数，并发送该函数的结果。
方法列表：
栗子：
Observable.just(2) .map(new Func1() { @Override public String call(Integer integer) { return String.valueOf(String.format("原始数据的两倍为: %s", integer * 2)); } });
Observable 对象为另一个 Observable 对象。
方法列表：
栗子：
Observable.just(2) .flatMap(new Func1>() { @Override public Observable call(Integer integer) { // 转换为一个定时 integer 秒的 Observable 对象 return Observable.timer(integer, TimeUnit.SECONDS); } });
public final Observable filter(Func1 super T, Boolean> predicate)
栗子：
Observable.just(-1, -2, 0, 1, 2) .filter(new Func1() { @Override public Boolean call(Integer integer) { return integer > 0; } });
Observable 发射的第一个［满足指定谓词的］元素的 Observable，如果如果源 Observable 为空，则会抛出一个 NoSuchElementException。
方法列表：
栗子：
// 发射第一个元素Observable.just(-1, -2, 0, 1, 2).first();// 发射满足条件的第一个元素Observable.just(-1, -2, 0, 1, 2) .first(new Func1() { @Override public Boolean call(Integer integer) { return integer > 0; } });// 会抛出 NoSuchElementException 异常Observable.empty().first();
Observable 发射的倒数第一个［满足指定谓词的］元素的 Observable，如果如果源 Observable 为空，则会抛出一个 NoSuchElementException。
方法列表：
栗子：
// 发射倒数第一个元素Observable.just(-1, -2, 0, 1, 2).first();// 发射满足条件的倒数第一个元素Observable.just(-1, -2, 0, 1, 2) .first(new Func1() { @Override public Boolean call(Integer integer) { return integer public final Observable
skip(int count) public final Observable
skip(long time, TimeUnit unit, Scheduler scheduler) 栗子：
Observable.just(-1, -2, 0, 1, 2) .skip(2) // 跳过前两条数据Observable 发射的任何数据。
方法列表：
栗子：
Observable.just(-1, -2, 0, 1, 2) .skip(2) // 跳过后两条数据public final Observable
take(final int count) public final Observable
take(long time, TimeUnit unit, Scheduler scheduler) 栗子：
Observable.just(-1, -2, 0, 1, 2).take(3); // 只发射前三条数据Observable 发射的任何数据。
方法列表：
public final Observable
takeLast(int count, long time, TimeUnit unit) public final Observable
takeLast(int count, long time, TimeUnit unit, Scheduler scheduler) public final Observable
takeLast(long time, TimeUnit unit, Scheduler scheduler) 栗子：
Observable.just(-1, -2, 0, 1, 2).takeLast(3); // 只发射后三条数据1Observable 发射的最后一条数据。
方法列表：
栗子：
Observable.interval(300, TimeUnit.MILLISECONDS) .sample(2, TimeUnit.SECONDS)public final Observable
elementAt(int index) 栗子：
Observable.just(-1, -2, 0, 1, 2).elementAt(2); // 发射索引为 2 的数据public final Observable
elementAtOrDefault(int index, T defaultValue) 栗子：
Observable.just(-1, -2, 0, 1, 2).elementAtOrDefault(9, -5); // 发射索引为 9的数据，若不存在，则发射 -5public final Observable
ignoreElements() 栗子：
Observable.just(-1, -2, 0, 1, 2).ignoreElements()public final Observable
distinct() public final Observable
distinct(Func1 super T, ? extends U> keySelector) 栗子：
// 直接过滤Observable.just(-1, -2, 0, 1, 2, 1).distinct();// 通过生成的 key 值过滤Observable.just(-1, -2, 0, 1, 2, 1).distinct(new Func1() { @Override public Integer call(Integer integer) { // 随机生成 key return integer * (int)(Math.random() * 10); } }); Observable 每产生结果后，如果在规定的间隔时间内没有产生新的结果，则发射这个结果，否则会忽略这个结果。该操作符会过滤掉发射速率过快的数据项。
方法列表：
public final Observable
debounce(long timeout, TimeUnit unit) public final Observable
debounce(long timeout, TimeUnit unit, Scheduler scheduler) public final Observable
debounce(Func1 super T, ? extends Observable> debounceSelector) 栗子：
Observableobservable = Observable.create(new Observable.OnSubscribe () { @Override public void call(Subscriber super Integer> subscriber) { try { //产生结果的间隔时间分别为100、200、300...900毫秒 for (int i = 1; i 该栗子产生结果为：依次打印5、6、7、8。
http://www.apkbus.com/blog-873055-76488.html
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/4684/viewspace-2813195/ ，如需转载，请注明出处，否则将追究法律责任。
