public void getUserPermission(FailableCallback<UserPermission> permissionCallback) {Thread thread = new Thread(()-> {try {UserPermission permission = // query a web servicepermissionCallback.accept(permission);} catch(Exception ex) {permission.error(ex);}});thread.start();
}public void getTodos(FailableCallback<List<Todo>> todosCallback) {Thread thread = new Thread(()-> {getUserPermission(new FailableCallback() {public void accept(UserPermission permission) {if(permission.isValid()) {try {List<Todo> todosFromWeb = // query a web serviceif(!todosCallbackInstance.isDisposed()) {if(syncWithUIThread()) {todosCallback.accept(todosFromWeb);}}} catch(Exception ex) {if(!todosCallbackInstance.isDisposed()) {if(syncWithUIThread()) {todosCallback.error(ex);}}}}}public void error(Exception ex) {// Oh no!}});});thread.start();
}

这是非常糟糕的编码，它可能会变得更糟，应该只显示一个可以用ReactiveX解决的例子。 这些问题通常被认为是回调地狱／大坑。

Observable<Todo> todoObservable = Observable.create(emitter -> {try {List<Todo> todos = getTodos();for (Todo todo : todos) {emitter.onNext(todo);}emitter.onComplete();} catch (Exception e) {emitter.onError(e);}
});

典型的Observable可能会发出无限数据，就像一个点击监听器一样，UI监听器是不可预测的，通常用户可能会点击按钮或其他UI小部件。通常终止成功或失败的类型Maybe<T>，Single<T>和 Completable。
      Maybe<T>对象是一种异步java.util.Optional从Java 8。

Maybe<List<Todo>> todoMaybe = Maybe.create(emitter -> {try {List<Todo> todos = getTodos();if(todos != null && !todos.isEmpty()) {emitter.onSuccess(todos); （1）}else {emitter.onComplete();   （2）}} catch (Exception e) {emitter.onError(e);   （3）}
});

Single<T>对象也可以被认为是承诺，在异步框架中也很受欢迎，并且类似于Maybe<T>对象，但只有没有onComplete()方法。
Completable对象与Single<T>对象非常相似，但没有返回值，因此也不具有类似其他类型的泛型。Completable对象也可以看作是反应式java.lang.Runnable对象。
      除了这些可观察类型的最流行的create()方法之外，还有更多的方便方法来创建这些类型之一。

• Observable.just（） - 允许在其他数据类型周围创建一个可观察的包装
• Observable.fromIterable（） - 接受一个java.lang.Iterable <T>，并在数据结构中按顺序排列它们的值
• Observable.fromArray（） - 获取一个数组，并在数据结构中按顺序排列它们的值
• Observable.fromCallable（） - 允许为java.util.concurrent.Callable <V>创建一个observable
• Observable.fromFuture（） - 允许为java.util.concurrent.Future创建一个observable
• Observable.interval（） - 在给定间隔内发出长对象的可观察值
• ...

Observable<Todo> todoObservable = Observable.create(emitter -> { ... });// Simply subscribe with a io.reactivex.functions.Consumer<T>, which will be informed onNext()
Disposable disposable = todoObservable.subscribe(t -> System.out.print(t));// Dispose the subscription when not interested in the emitted data any more
disposable.dispose();// Also handle the error case with a second io.reactivex.functions.Consumer<T>
Disposable subscribe = todoObservable.subscribe(t -> System.out.print(t), e -> e.printStackTrace());// ...

DisposableObserver<Todo> disposableObserver = todoObservable.subscribeWith(new         DisposableObserver<Todo>() {@Overridepublic void onNext(Todo t) {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onComplete() {}
});

4、处理订阅并使用CompositeDisposable

当注册者或订阅者被附加时，他们通常不应该永久地聆听。所以可能会发生这样的情况：由于某些状态的改变，一个可观察的发射的事件可能不再是有趣的。

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableSingleObserver;Single<List<Todo>> todosSingle = getTodos();Disposable disposable = todosSingle.subscribeWith(new DisposableSingleObserver<List<Todo>>() {@Overridepublic void onSuccess(List<Todo> todos) {// work with the resulting todos}@Overridepublic void onError(Throwable e) {// handle the error case}
});// continue working and dispose when value of the Single is not interesting any more
disposable.dispose();

• 单一类和其他可观察类提供不同的订阅方法，返回一个Disposable对象。

Single<List<Todo>> todosSingle = Single.create(emitter -> {Thread thread = new Thread(() -> {try {List<Todo> todosFromWeb = // query a webserviceSystem.out.println("Called 4 times!");emitter.onSuccess(todosFromWeb);} catch (Exception e) {emitter.onError(e);}});thread.start();
});todosSingle.subscribe(... " Show todos times in a bar chart " ...);showTodosInATable(todosSingle);todosSingle.subscribe(... " Show todos in gant diagram " ...);anotherMethodThatsSupposedToSubscribeTheSameSingle(todosSingle);

Single<List<Todo>> todosSingle = Single.create(emitter -> {Thread thread = new Thread(() -> {try {List<Todo> todosFromWeb = // query a webserviceSystem.out.println("I am only called once!");emitter.onSuccess(todosFromWeb);} catch (Exception e) {emitter.onError(e);}});thread.start();
});// cache the result of the single, so that the web query is only done once
Single<List<Todo>> cachedSingle = todosSingle.cache();cachedSingle.subscribe(... " Show todos times in a bar chart " ...);showTodosInATable(cachedSingle);cachedSingle.subscribe(... " Show todos in gant diagram " ...);anotherMethodThatsSupposedToSubscribeTheSameSingle(cachedSingle);

6. Flowable<T> 和 Backpressure

RxJava 2.0引入了一种新型的Flowable <T>，它与API相当于Observable <T>，但Flowable <T>支持Backpressure，而Observable <T>则不支持。回到RxJava 1.0Backpressure的概念太晚了，被添加到了Observable的类型，但有些则抛出了一个MissingBackpressureException，所以区分Flowable <T>和Observable <T>是一件好事。除了Observable <T>也可能<T>，Single <T>和Completable没有Backpressure。

7.类型之间的转换

很容易在不同的RxJava类型之间进行转换。

8、测试RxJava可观察和订阅

8.1、 测试可观察量

您可以通过RxJava库提供的TestSubscriber类来测试可观察值。

Observable<String> obs = ...// assume creation code here
TestSubscriber<String> testSubscriber = new TestSubscriber<>();
obs.subscribe(testSubscriber);testSubscriber.assertNoErrors();
List<String> chickens = testSubscriber.getOnNextEvents();// TODO assert your string integrity...

8.2、 测试可观察量

RxJava提供了一种覆盖显示的调度程序的方法，以便可观察器被同步调用。 请参阅http://fedepaol.github.io/blog/2015/09/13/testing-rxjava-observables-subscriptions/作为示例。