在Android中运用RxJava_import io.reactivex.single maven

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