参考 http://gank.io/post/560e15be2dca930e00da1083
Rxjava的基本用法：
基础使用
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> o) {
subscriber.onNext("hello,Rx");
subscriber.onCompleted();
}
});
Observer observer = new Observer() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
}
};
observable.subscribe(observer);
Observable observable = Observable.create(new onSubscribe);
public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
}
public interface Action1<T> extends Action {
void call(T t);
}
public interface Action extends Function {
}
public class Observable<T> {
final OnSubscribe<T> onSubscribe;
private static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
public final static <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(hook.onCreate(f));
}
protected Observable(OnSubscribe<T> f) {
this.onSubscribe = f;
}
}
public abstract class RxJavaObservableExecutionHook {
public <T> OnSubscribe<T> onCreate(OnSubscribe<T> f) {
return f;
}
Observable类里存在成员变量onSubscribe。
Observable.create方法的参数是OnSubscribe的实现类。方法里调用了构造方法Observable(OnSubscribe)。参数是hook.oncreate(OnSubscribe)的返回值，追入hook类，hook.onCreate(OnSubscribe)方法返回传入的onSubscribe，所以构造方法的参数就是我们传入的onSubscribe对象。
所以，create方法的作用是创建一个Observable，并将传入的onSubscribe赋值给自己的成员变量。
Observer observer = new Observer() ；
public interface Observer<T> {
void onCompleted();
void onError(Throwable e);
void onNext(T t);
}
创建一个Observer接口的实现类。
observable.subscribe(observer);
public final Subscription subscribe(final Observer<? super T> observer) {
if (observer instanceof Subscriber) {
return subscribe((Subscriber<? super T>)observer);
}
return subscribe(new Subscriber<T>() {
@Override
public void onCompleted() {
observer.onCompleted();
}
@Override
public void onError(Throwable e) {
observer.onError(e);
}
@Override
public void onNext(T t) {
observer.onNext(t);
}
});
}
Observable类的subscribe方法是将传入的Observer对象封装成Subscriber对象。
public abstract class Subscriber<T> implements Observer<T>, Subscription {
private static final Long NOT_SET = Long.MIN_VALUE;
private final SubscriptionList subscriptions;
private final Subscriber<?> subscriber;
private Producer producer;
private long requested = NOT_SET; // default to not set
protected Subscriber() {
this(null, false);
}
//...其他构造方法和成员方法
public final void add(Subscription s) {
subscriptions.add(s);
}
@Override
public final void unsubscribe() {
subscriptions.unsubscribe();
}
@Override
public final boolean isUnsubscribed() {
return subscriptions.isUnsubscribed();
}
public void onStart() {
// 空实现
}
}
public interface Subscription {
void unsubscribe();
boolean isUnsubscribed();
}
subscriber实现了Observer和Subscription的抽象类，Subcriptions暂且用不到，我也不知道是干啥的。
Observable的subscribe方法将传入的Observer对象封装成Subscriber对象之后，调用subscribe(Subscriber)方法,且传入的第二个参数是Observable本身。
public final Subscription subscribe(Subscriber<? super T> subscriber) {
return Observable.subscribe(subscriber, this);
}
调用重载方法subscribe(Subscriber, Observable)
private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
if (subscriber == null) {
throw new IllegalArgumentException("observer can not be null");
}
if (observable.onSubscribe == null) {
throw new IllegalStateException("onSubscribe function can not be null.");
}
subscriber.onStart();
if (!(subscriber instanceof SafeSubscriber)) { //包装成安全的subscriber
subscriber = new SafeSubscriber<T>(subscriber);
}
try {
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
try {
subscriber.onError(hook.onSubscribeError(e));
} catch (Throwable e2) {
Exceptions.throwIfFatal(e2);
RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
hook.onSubscribeError(r);
throw r;
}
return Subscriptions.unsubscribed();
}
}
方法里首先判空，然后调用了subscriber的onstart方法，再将subscriber对象包装成安全的subscriber，调用hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {
// pass-thru by default
return onSubscribe;
}
hook.onSubscribeStart返回observable.onSubcribe对象。调用call()方法并将subscriber作为参数传入。call方法是在第一步创建Observable对象时实现的。hook.onSubscribeReturn(subscriber)返回subscriber对象。
Just && From
Observable.just("hello").subscribe(observer);
Observable.from(new String[]{"hello","Rx"}).subscribe(observer);
just
public final static <T> Observable<T> just(final T value) {
return ScalarSynchronousObservable.create(value);
}
public static final <T> ScalarSynchronousObservable<T> create(T t) {
return new ScalarSynchronousObservable<T>(t);
}
protected ScalarSynchronousObservable(final T t) {
super(new OnSubscribe<T>() {
@Override
public void call(Subscriber<? super T> s) {
s.onNext(t);
s.onCompleted();
}
});
this.t = t;
}
from
public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
return create(new OnSubscribeFromIterable<T>(iterable));
}
public final class OnSubscribeFromIterable<T> implements OnSubscribe<T> {
final Iterable<? extends T> is;
public OnSubscribeFromIterable(Iterable<? extends T> iterable) {
if (iterable == null) {
throw new NullPointerException("iterable must not be null");
}
this.is = iterable;
}
@Override
public void call(final Subscriber<? super T> o) {
final Iterator<? extends T> it = is.iterator();
if (!it.hasNext() && !o.isUnsubscribed())
o.onCompleted();
else
// 调用subscriber的setProducer方法，内部是迭代Iterator获取参数并调用onNext方法
o.setProducer(new IterableProducer<T>(o, it));
}
}
public void setProducer(Producer p) {
long toRequest;
boolean passToSubscriber = false;
synchronized (this) {
toRequest = requested;
producer = p;
if (subscriber != null) { //假
if (toRequest == NOT_SET) {
// we pass-thru to the next producer as nothing has been requested
passToSubscriber = true;
}
}
}
if (passToSubscriber) { //假
subscriber.setProducer(producer);
} else {
if (toRequest == NOT_SET) {
producer.request(Long.MAX_VALUE); //走这里 调用IterableProducer的request方法
} else {
producer.request(toRequest);
}
}
}
private static final class IterableProducer<T> extends AtomicLong implements Producer {
public void request(long n) {
if (get() == Long.MAX_VALUE) { // get是AtomicLong的方法。此处是0。
return;
}
// 比较对象的Value是否等于0，如果是就为Value赋值Long.MAX_VALUE并返回true。此处为true
// 调用fastpath();
if (n == Long.MAX_VALUE && compareAndSet(0, Long.MAX_VALUE)) {
fastpath();
} else
if (n > 0 && BackpressureUtils.getAndAddRequest(this, n) == 0L) {
slowpath(n);
}
}
}
// 迭代调用onNext方法
void fastpath() {
final Subscriber<? super T> o = this.o;
final Iterator<? extends T> it = this.it;
while (true) {
if (o.isUnsubscribed()) {
return;
} else if (it.hasNext()) {
o.onNext(it.next());
} else if (!o.isUnsubscribed()) {
o.onCompleted();
return;
} else {
// is unsubscribed
return;
}
}
}
ActionX
Action1<String> action1 = new Action1<String>() {
@Override
public void call(String str) {
Log.d(tag,str);
}
};
Observable.just("hello").subscribe(action1);
以上是简单用法。ActionX的X代表call方法里参数的个数。
public final Subscription subscribe(final Action1<? super T> onNext) {
if (onNext == null) {
throw new IllegalArgumentException("onNext can not be null");
}
return subscribe(new Subscriber<T>() {
@Override
public final void onCompleted() {
// do nothing
}
@Override
public final void onError(Throwable e) {
throw new OnErrorNotImplementedException(e);
}
@Override
public final void onNext(T args) {
onNext.call(args);
}
});
}
subscribe方法把Action1包装成一个Subscriber对象。并且在onNext方法里调用action1的call方法。
类似的方法有：
subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError)
subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete)