RxJava的使用与深入学习_rxjava 使用

简单使用（以Android平台为例）

compile 'io.reactivex:rxjava:1.1.3'
compile 'io.reactivex:rxandroid:1.1.0'

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("Hi");
        subscriber.onNext("Aloha");
        subscriber.onCompleted();
    }
});
Observable observable = Observable.just("Hello", "Hi", "Aloha");
String[] words = {"Hello", "Hi", "Aloha"};  Observable observable = Observable.from(words);

Subscriber subscriber = new Subscriber() {
            @Override
            public void onCompleted() {
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onNext(Object o) {
            }
            @Override
            public void onStart() {
                super.onStart();
            }
};

observable.subscribe(observer); 
observable.subscribe(subscriber);

Observable.just(1, 2, 3, 4) 
    .subscribeOn(Schedulers.io()) //决定调用observable.subscribe(subscriber)方法时的执行线程
    .observeOn(Schedulers.newThread()) //决定下面mapOperator方法的执行线程
    .map(mapOperator) 
    .observeOn(Schedulers.io()) //决定下面mapOperator2方法的执行线程
    .map(mapOperator2) 
    .observeOn(AndroidSchedulers.mainThread) //决定下面subscriber对象的onNext()、onCompleted()、onCompleted()和onError()方法执行线程。即UI线程
    .subscribe(subscriber);  

• Observable对象创建的三个方法create、just、from；
• observable.subscribe(observer)内部逻辑是什么，为何会最终执行到Subscriber对象的相关方法；
• Observable中Observable.OnSubscribe对象的call方法何时被调用；
• Subscriber 对象的onStart、unSubscribe、onNext()、onCompleted()、onCompleted()和onError()在何处，何种情况被调用；
• Observable的map和flatMap方法内部实现机制，参数的传递；
• Schedulers如何创建指定的线程，各个线程之间的区别是什么；
• Observable的observeOn和subscribeOn方法的内部实现机制，如何实现线程的切换；

final OnSubscribe<T> onSubscribe;
static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
//RxJavaPlugins.getInstance()是一个单例模式，获取一个RxJavaPlugins对象，然后这个对象定义了Hook，然而这对Hook并没有做任何事情，至少目前是这样的。
//rxJavaPlugins.getObservableExecutionHook()近似等价于得到一个RxJavaObservableExecutionHook对象，该对象目前不干任何事，完全可以把它看成透明的，你传什么进去它回什么。

protected Observable(OnSubscribe<T> f) {
        this.onSubscribe = f;
}

public static <T> Observable<T> create(OnSubscribe<T> f) {
        return new Observable<T>(hook.onCreate(f));
}

public static <T> Observable<T> just(final T value) {
        return ScalarSynchronousObservable.create(value); //note1
}
public static <T> Observable<T> just(T t1, T t2) {
        return from((T[])new Object[] { t1, t2 }); //note2
}

public static <T> Observable<T> from(T[] array) { 
        int n = array.length;
        if (n == 0) {
            return empty();//note1
        } else
        if (n == 1) {
            return just(array[0]);//note2
        }
        return create(new OnSubscribeFromArray<T>(array));
}

private static final class EmptyHolder {
        final static Observable<Object> INSTANCE = create(new OnSubscribe<Object>() {
            @Override
            public void call(Subscriber<? super Object> subscriber) {
                subscriber.onCompleted();
            }
        });
}

• 方法只含一个参数：调用ScalarSynchronousObservable.create(value)创建一个ScalarSynchronousObservable对象。
• 方法含多个参数：首先将多个参数构造成一个T[] array对象数组，随后利用该数组创建一个OnSubscribeFromArray<T>(array)的对象，最后调用Observable(OnSubscribe<T> f)方法创建一个Observable对象。

public final Subscription subscribe(final Observer<? super T> observer) {
        if (observer instanceof Subscriber) {
            return subscribe((Subscriber<? super T>)observer); //note1
        }
        return subscribe(new Subscriber<T>() { //note2
            @Override
            public void onCompleted() {
                observer.onCompleted();
            }
            @Override
            public void onError(Throwable e) {
                observer.onError(e);
            }
            @Override
            public void onNext(T t) {
                observer.onNext(t);
            }
        });
}

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();//note1
        if (!(subscriber instanceof SafeSubscriber)) {
            subscriber = new SafeSubscriber<T>(subscriber); 
        }
        try {
            hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);//note2
            return hook.onSubscribeReturn(subscriber);//note3
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            try {
                subscriber.onError(hook.onSubscribeError(e)); //note4
            } 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();
        }
}

public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
        return lift(new OperatorMap<T, R>(func));
}

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return new Observable<R>(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> o) {
                try {
                    Subscriber<? super T> st = hook.onLift(operator).call(o); //note1
                    try {
                        st.onStart(); //note2
                        onSubscribe.call(st); //note3
                    } catch (Throwable e) {
                        Exceptions.throwIfFatal(e);
                        st.onError(e); //note4
                    }
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    o.onError(e); //note5
                }
            }
        });
}

Observable.just("images/logo.png") //note1 
    .map(new Func1<String, Bitmap>() {
        @Override
        public Bitmap call(String filePath) {
            return getBitmapFromPath(filePath); 
        }
    }) //note2
    .subscribe(new Action1<Bitmap>() { //note3
        @Override
        public void call(Bitmap bitmap) { 
            showBitmap(bitmap);
        }
});

• Func1<T,R>将T类型的数据转换为R类型数据；
• OperatorMap<T, R>(func) 将Subscriber<R>转换成一个Subscriber<T>，前者Subscriber接受参数为R，后者接收参数为T
• lift方法创建一个Observable<R>对象，对应一个OnSubscribe<R>，其call方法参数Subscriber<R>，表明Subscriber.onNext方法接受参数类型为R。

Observable.just("images/logo.png") // 对应Observable<String>
    .map(new Func1<String, T1>() { })  // 对应Observable<T1>
    .map(new Func1<T1, T2>() { })  // 对应Observable<T2>
    .map(new Func1<T2, R>() { })  // 对应Observable<R> 
    .subscribe(new Action1<R>() {  // 对应Observable<R> 
        @Override
        public void call(R r) { 
            //
        };

Observable.just("images/logo.png") // 对应Observable<String>
    .subscribe(new Action1<R>() {  // 对应Observable<R> 
        @Override
        public void call(R new Func1<T1, T2>().call (new Func1<T1, T2>().call(new Func1<String, T1>().call(String) )  ) ) { 
};

public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler, RxRingBuffer.SIZE);
}
public final Observable<T> observeOn(Scheduler scheduler, int bufferSize) {
        return observeOn(scheduler, false, bufferSize);
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) { //note1
        if (this instanceof ScalarSynchronousObservable) {
            return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler); //note2
        }
        return lift(new OperatorObserveOn<T>(scheduler, delayError, bufferSize)); //note3
}

public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
        try {
            subscriber.onStart(); //note1
            onSubscribe.call(subscriber); //note2
            return subscriber;
        } catch (Throwable e) {
            try {
                subscriber.onError(e);
            } catch (Throwable e2) {
               ...
            }
            return Subscriptions.unsubscribed();
        }
}

final Subscriber<? super T> child;
final Scheduler.Worker recursiveScheduler;
final NotificationLite<T> on;
final Queue<Object> queue;
volatile boolean finished; //当前流的状态
long emitted;//当前被处理的事件数
final int limit; //处理的门限，达到该值需要重新调用request方法
final AtomicLong requested = new AtomicLong();
final AtomicLong counter = new AtomicLong();

public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child, boolean delayError, int bufferSize) {
            this.child = child; //note1
            this.recursiveScheduler = scheduler.createWorker(); //note2
            this.on = NotificationLite.instance(); //note3
            this.limit = calculatedSize - (calculatedSize >> 2);
            int calculatedSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;
            if (UnsafeAccess.isUnsafeAvailable()) { //note4
                queue = new SpscArrayQueue<Object>(calculatedSize);
            } else {
                queue = new SpscAtomicArrayQueue<Object>(calculatedSize); //note5
            }
            request(calculatedSize); //note6
}

void init() {
            Subscriber<? super T> localChild = child;
            localChild.setProducer(new Producer() { //note0
                @Override
                public void request(long n) {
                    if (n > 0L) {
                        BackpressureUtils.getAndAddRequest(requested, n); //note1
                        schedule(); //note2
                    }
                }
            });
            localChild.add(recursiveScheduler); //note3
            localChild.add(this); //note4
}

protected void schedule() {
            if (counter.getAndIncrement() == 0) {//note1
                recursiveScheduler.schedule(this); //note2
            }
}

@Override  public void onNext(final T t) { 
            if (isUnsubscribed() || finished) {
                return;
            }
            if (!queue.offer(on.next(t))) { //note1
                onError(new MissingBackpressureException());
                return;
            }
            schedule();//note2
}

@Override  public void onCompleted() { 
            if (isUnsubscribed() || finished) {
                return;
            }
            finished = true; //note1
            schedule(); //note2
}

@Override  public void onError(final Throwable e) { 
            if (isUnsubscribed() || finished) {
                RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
                return;
            }
            error = e; //ntoe1
            finished = true; //note1
            schedule(); //note2
}

public abstract Worker createWorker();//note1
public abstract static class Worker implements Subscription {
    public abstract Subscription schedule(Action0 action);//note2
    public abstract Subscription schedule(final Action0 action, final long delayTime, final TimeUnit unit);
}

private static final String THREAD_NAME_PREFIX = "RxNewThreadScheduler-";//note1
private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);//note2
public Worker createWorker() {
        return new NewThreadWorker(THREAD_FACTORY); //note3
}

public NewThreadWorker(ThreadFactory threadFactory) {
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory); //note1
        ....
        executor = exec;
}

public Subscription schedule(final Action0 action) {
        return schedule(action, 0, null);
}
public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) { 
        if (isUnsubscribed) {//note2
            return Subscriptions.unsubscribed();
        }
        return scheduleActual(action, delayTime, unit);
}

public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
        ScheduledAction run = new ScheduledAction(action); //note1
        Future<?> f;
        if (delayTime <= 0) {
            f = executor.submit(run); //note2
        } else {
            f = executor.schedule(run, delayTime, unit);
        }
        return run;
}

public static Scheduler mainThread() {
        Scheduler scheduler =  RxAndroidPlugins.getInstance().getSchedulersHook().getMainThreadScheduler(); //note1
        return scheduler != null ? scheduler : MainThreadSchedulerHolder.MAIN_THREAD_SCHEDULER; //note2
}

private final Handler handler;
HandlerScheduler(Handler handler) {
        this.handler = handler;
}

public Worker createWorker() {
        return new HandlerWorker(handler);
}
static class HandlerWorker extends Worker {
        private final Handler handler;
        HandlerWorker(Handler handler) {
            this.handler = handler;
        }
        @Override
        public Subscription schedule(final Action0 action) {
            return schedule(action, 0, TimeUnit.MILLISECONDS);
        }
        @Override
        public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
            if (compositeSubscription.isUnsubscribed()) {//note0
                return Subscriptions.unsubscribed();
            }
            action = RxAndroidPlugins.getInstance().getSchedulersHook().onSchedule(action); //note1
            final ScheduledAction scheduledAction = new ScheduledAction(action); //note2
            ...
            handler.postDelayed(scheduledAction, unit.toMillis(delayTime)); //note3
            ....
            return scheduledAction;
        }
        
}

final T t;
protected ScalarSynchronousObservable(final T t) {
        super(new OnSubscribe<T>() {
            @Override
            public void call(Subscriber<? super T> s) {
                s.setProducer(createProducer(s, t)); //note1
            }
        });
        this.t = t;
}

static <T> Producer createProducer(Subscriber<? super T> s, T v) {
        if (STRONG_MODE) { //默认是false
            return new SingleProducer<T>(s, v);
        }
        return new WeakSingleProducer<T>(s, v); 返回这个Producer
}

static final class WeakSingleProducer<T> implements Producer{
    final Subscriber<? super T> actual;
    final T value;
    boolean once;
    public void request(long n) {
            if (once) {
                return;
            }
            if (n < 0L) {
                throw new IllegalStateException("n >= required but it was " + n);
            }
            if (n != 0L) {
                once = true;
                Subscriber<? super T> a = actual;
                if (a.isUnsubscribed()) {
                    return;
                }
                T v = value;
                try {
                    a.onNext(v); //note1
                } catch (Throwable e) {
                    Exceptions.throwOrReport(e, a, v);
                    return;
                }
                if (a.isUnsubscribed()) {
                    return;
                }
                a.onCompleted(); //note2
            }
        }
} 

public <R> Observable<R> scalarFlatMap(final Func1<? super T, ? extends Observable<? extends R>> func) {
        return create(new OnSubscribe<R>() {
            @Override  public void call(final Subscriber<? super R> child) { 
                Observable<? extends R> o = func.call(t); //note1
                if (o instanceof ScalarSynchronousObservable) {
                    child.setProducer(createProducer(child, ((ScalarSynchronousObservable<? extends R>)o).t)); //note2
                } else {
                    o.unsafeSubscribe(child); //note3
                }
            }
        });
}

@Override public void call(Subscriber<? super T> child) { 
        child.setProducer(new FromArrayProducer<T>(child, array));
}

static final class FromArrayProducer<T>  extends AtomicLong implements Producer
final Subscriber<? super T> child;
final T[] array;
int index;
public FromArrayProducer(Subscriber<? super T> child, T[] array) {
            this.child = child;
            this.array = array;
}
@Override  public void request(long n) { 
            if (n < 0) {
                throw new IllegalArgumentException("n >= 0 required but it was " + n);
            }
            if (n == Long.MAX_VALUE) {
                if (BackpressureUtils.getAndAddRequest(this, n) == 0) { //note1
                    fastPath();
                }
            } else   if (n != 0) { 
                if (BackpressureUtils.getAndAddRequest(this, n) == 0) {
                    slowPath(n);
                }
            }
}

void fastPath() {
            final Subscriber<? super T> child = this.child;
            for (T t : array) {
                if (child.isUnsubscribed()) {
                    return;
                }
                child.onNext(t); //note1
            }
            if (child.isUnsubscribed()) {
                return;
            }
            child.onCompleted(); //note2
}

void slowPath(long r) {
            final Subscriber<? super T> child = this.child;
            final T[] array = this.array;
            final int n = array.length;
            long e = 0L;
            int i = index; //若是第一次执行这里的index=0
            for (;;) {
                while (r != 0L && i != n) { //note1
                    if (child.isUnsubscribed()) {
                        return;
                    }
                    child.onNext(array[i]);
                    i++;
                    if (i == n) {
                        if (!child.isUnsubscribed()) {
                            child.onCompleted();
                        }
                        return;
                    }
                    r--;
                    e--;
                } //note2
                r = get() + e; //note3
                if (r == 0L) { //note2
                    index = i; //记录当前执行到的任务数
                    r = addAndGet(e);
                    if (r == 0L) {
                        return;
                    }
                    e = 0L;
                }
            }
        }

private final SubscriptionList subscriptions; //构造器中赋初值
private final Subscriber<?> subscriber; //构造器中赋初值
private Producer producer;  //事件制造者
private long requested = NOT_SET; //最大接受任务数
private static final Long NOT_SET = Long.MIN_VALUE;

protected Subscriber() {  this(null, false); }
protected Subscriber(Subscriber<?> subscriber) { this(subscriber, true); }
protected Subscriber(Subscriber<?> subscriber, boolean shareSubscriptions) {
        this.subscriber = subscriber;
        this.subscriptions = shareSubscriptions && subscriber != null ? subscriber.subscriptions : new SubscriptionList(); //note1
}

public final void add(Subscription s) {
        subscriptions.add(s);
}

<pre name="code" class="java">public void onStart() {
        // do nothing by default
}

@Override public final void unsubscribe() {
        subscriptions.unsubscribe();
}

@Override public final boolean isUnsubscribed() {
        return subscriptions.isUnsubscribed();
}

protected final void request(long n) {
        if (n < 0) {
            throw new IllegalArgumentException("number requested cannot be negative: " + n);
        }
        Producer producerToRequestFrom = null;
        synchronized (this) {
            if (producer != null) {
                producerToRequestFrom = producer;
            } else {
                addToRequested(n); //note1
                return;
            }
        }
        producerToRequestFrom.request(n); //note2
}

private void addToRequested(long n) {
        if (requested == NOT_SET) {
            requested = n;
        } else {
            final long total = requested + n;
            if (total < 0) {
                requested = Long.MAX_VALUE; //note1
            } else {
                requested = total;
            }
        }
    }

public void setProducer(Producer p) {
        long toRequest;
        boolean passToSubscriber = false;
        synchronized (this) {
            toRequest = requested; //note1
            producer = p; 
            if (subscriber != null) {//一般情况下该结果为假
                if (toRequest == NOT_SET) {
                    passToSubscriber = true;
                }
            }
        }
        if (passToSubscriber) {//一般情况该行结果为假
            subscriber.setProducer(producer);
        } else {
            if (toRequest == NOT_SET) { //note2
                producer.request(Long.MAX_VALUE);
            } else {
                producer.request(toRequest);
            }
        }
 

1. 利用operator将subscriber<R>转成subscriber<T>  [或者将subscriber_inner<T>转成subscriber<T> ], Operator在此起到了连接subscriber<T>和subscriber<R>的作用，可以对来自subscriber<T>的t数据进行处理之后再传递给subscriber<R>
2. 调用subscriber<T>对象的onstart方法, 
3. 调用observable<T>.onsubscribe.call方法，
   • call方法内部会调用subscriber<T>的onNext 、onCompleted方法。onNext方法内部先使用Operator进行一定的预处理，之后根据预处理的结果执行Subscribe<R>.onNext方法
   • 以map为例：subscribe<T>.onnext(T t) 内部执行过程是{ subscribe<R>.onnext(Fun1.call(t)) }
   • 以observeOn为例：subscribe<T>.onnext(T t) 内部执行过程是{queue.add(t); executor.submit(new Runnable(){  call() }); } 上面的call方法内容如下{ t=queue.get();  subscribe_inner<T>.onnext(t);}

import rx.Observable;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;
public class RxBus {
    private final Subject<Object, Object> _bus = new SerializedSubject<>(PublishSubject.create());
    public void send(Object o) {  _bus.onNext(o);  } 
    public Observable<Object> toObserverable() {  return _bus;  } 
    public boolean hasObservers() {  return _bus.hasObservers();  } 
}

public class SerializedSubject<T, R> extends Subject<T, R> {
    private final SerializedObserver<T> observer;
    private final Subject<T, R> actual;
    public SerializedSubject(final Subject<T, R> actual) {
        super(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> child) {
                actual.unsafeSubscribe(child);   
            }
        });
        this.actual = actual;
        this.observer = new SerializedObserver<T>(actual);
    }
    @Override  public void onCompleted() {  observer.onCompleted(); } 
    @Override  public void onError(Throwable e) { observer.onError(e);  } 
    @Override   public void onNext(T t) {  observer.onNext(t); } 
    @Override public boolean hasObservers() {   return actual.hasObservers();  } 
}

与EventBus的对比