public void getUserPermission(FailableCallback<UserPermission> permissionCallback) {
        Thread thread = new Thread(()-> {
                try {
                        UserPermission permission = // query a web service

                        permissionCallback.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 service

                                                if(!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();
}

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);
        }
});

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）
        }
});

• 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>() {

        @Override
        public void onNext(Todo t) {
        }

        @Override
        public void onError(Throwable e) {
        }

        @Override
        public 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>>() {

        @Override
        public void onSuccess(List<Todo> todos) {
                // work with the resulting todos
        }

        @Override
        public 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 webservice

                        System.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 webservice

                        System.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

7.类型之间的转换

8、测试RxJava可观察和订阅

8.1、 测试可观察量

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、 测试可观察量