RxJava 2.0配置
在项目Build.gradle文件里面添加如下代码，即可：
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
基本操作
1.Observable.create()
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
for(int i=0; i<5; i++){
e.onNext(String.valueOf(i));
}
e.onComplete();
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
a). 传入ObservableOnSubscribe 里面就一个方法subscribe，传入一个可观察的数据发射器，继承关系：ObservableEmitter -> Emitter。Emitter代表发射器，代码如下：
public interface Emitter<T> {
void onNext(T value);
void onError(Throwable error);
void onComplete();
}
这和RxJava1.0使用的Subscriber的主要功能是一模一样的，提供数据源的发送。
public interface Subscriber<T> {
public void onSubscribe(Subscription s);
public void onNext(T t);
public void onError(Throwable t);
public void onComplete();
}
b).我们常用的subscribe方法列表，如下：
Disposable subscribe(Consumer<? super T> onNext)
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete， Consumer<? super Disposable> onSubscribe)
void subscribe(Observer<? super T> observer)
Disposable定义如下，与RxJava1.0 Subscription对应，提供流的解除订阅：
public interface Disposable {
void dispose();
boolean isDisposed();
}
Observer定义如下：
public interface Observer<T> {
void onSubscribe(Disposable d);
void onNext(T t);
void onError(Throwable e);
void onComplete();
}
其中，我们将看不到RxJava1.0的action1之类的接口，取而代之的是与Java8命名类似的函数式接口。
查看源码，subscribe前几个方法最后都是调用的都是subscribe(Observer observer)，如下：
public Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
Action onComplete, Consumer<? super Disposable> onSubscribe) {
...
LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
subscribe(ls);
return ls;
}
LambdaObserver简单封装了我们传入的函数接口，作为一个Observer。
我们再看最基本的方法subscribe(Observer)的实现，如下：
public final void subscribe(Observer<? super T> observer) {
observer = RxJavaPlugins.onSubscribe(this, observer);
subscribeActual(observer);
}
RxJavaPlugins就是以前RxJava的RxJavaHooks，可以看成代理包装，不设置就是直接返回原对象，可以忽略，最后调用subscribeActual（）这个是抽象方法，真正调用流都是在这个方法里面进一步实现的。
2.分析Observable.just()调用
示例代码，打印“5”，如下：
Disposable d = Observable.just("5").subscribe(s -> System.out.println(s));
Observable.just 是最简单的rxjava操作了，就是生成数据，如下：
public static <T> Observable<T> just(T item) {
return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
}
很简单，生成一个ObservableJust对象，内容如下：
public final class ObservableJust<T> extends Observable<T> implements ScalarCallable<T> {
private final T value;
public ObservableJust(final T value) {
this.value = value;
}
@Override
protected void subscribeActual(Observer<? super T> s) {
ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
s.onSubscribe(sd);
sd.run();
}
@Override
public T call() {
return value;
}
}
所以，我们调用Observable.just()，会拿到我们调用链的第一个Observable对象---ObservableJust，它继承了Observable<T>并实现subscribeActual()方法。如上面示例代码，根据前面分析，我们将打印的Consumer-->转换为了LambdaObserver，传入了ObservableJust对象的subscribeActual（LambdaObserver）。
下一步，传入obersver和value，生成ScalarDisposable对象，直接调用该对象的run方法，完成了整个调用过程。ScalarDisposable.run方法，如下：
@Override
public void run() {
if (get() == START && compareAndSet(START, ON_NEXT)) {
observer.onNext(value);
if (get() == ON_NEXT) {
lazySet(ON_COMPLETE);
observer.onComplete();
}
}
}
忽略其他代码，基本看出，调用了我们传入的observer.onNext() -> observer.onComplete()，完成打印。
3.组合Observable.just() 和 Observable.map()
示例代码，将数字转换为字符串，并打印，如下：
Disposable d = Observable.just(1).map(i -> String.valueOf(i)).subscribe(s -> System.out.println(s));
我们知道，Observable.just()返回了一个ObservableJust对象，也就是Observable<T>的实例，因此，上述等式相当于，如下：
new ObservableJust().map(xxx);
因此我们先记住，当前对象是ObservableJust，打开map方法定义，如下：
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
可以看到，返回了一个ObservableMap对象，该对象传入了一个this，也就是当前的Observable对象，所以我们可以知道，如果这个操作符是转换用的，肯定会传入this当前observable对象，而Just操作是数据源的开头，所以不需要。ObservableMap代码如下：
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
}
一样的我们点开AbstractObservableWithUpstream<T, U>知道它继承Observable<U>对象，转换流的能力抽象，从原始的T类型流，转换为我们需要的U类型流。ObservableMap的构造函数传入了原始的T类型流，也就是我们的ObservableJust<Integer>类型流，以及我们的转换函数接口，将subscribeActual具体实现的时候，ObservableJust收到的Observer对象是个类似Observer<Integer>代理对象，里面封装了map的转换逻辑和原始的observer<String>对象，到这里ObservableJust的数据就全部接入MapObserver里面。MapObserver代码如下：
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}
@Override
public void onNext(T t) {
...
U v;
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper
actual.onNext(v);
}
}
代码可以看出，收到ObservableJust对象的Integer值后，通过转换函数接口Function，将T类型转换了U类型，也就是我们要的String类型，然后发送给实际的接收者，完成整个流过程。伪代码如下：
方法组合过程：
ObservableJust just = new ObservableJust(数据);
ObservableMap map = new ObservableMap (just, function);
订阅过程：
Observer 实际接收者；
--->
map.subscribe(实际接收者)；
--->
Observer mapObserver = new Observer(实际接收者);
just.subsribe(mapObserver);
--->
导致数据流动开始，int ---> mapObserver --> string --> 实际接收者
简单分析了一下组合调用流程。
网友评论