RxJava响应式编程学习笔记

1、概述

RxJava是一个著名的开源库，是ReactiveX(Reactive Extensions)的一种java实现。ReactiveX是一种响应式扩展框架，有很多实现，如RxAndroid,RxJS,RxSwift,RxRuby,RxCpp,RxGo等。RxJava有1.x和2.x两个主要的分支，分别代表着RxJava1和RxJava2。

RxJava可以看作由Observable,Subscriber和Scheduler组成，它们关系图为

Subscriber订阅到Observable，Observable会在默认或者指定的Scheduler上工作并产生数据流返回给Subscriber，Subscriber也会在默认或者在指定的Scheduler上接收Observable要送过来的数据流。Scheduler是对线程的抽象，不同的Scheduler代表了不同的线程

1.1  Observable和Subscriber

Observable提供了subscribe方法，当有Subscriber通过subscribe方法订阅到Observable时，Observable可以向Scheduler发送数据流。响应式编程中的事件分为三类：普通事件、错误事件和结束事件。在Subscriber中有三个方法与这三种事件一个一个地对应。Observable会通过调用Subscriber的这三个方法发送对应的事件。

• onNext:当Observable要发送普通事件时，就会调用这个方法，可以被调用0-N次
• onError:当在Observable内部有异常或者错误产生的时候，可以调用这个方法来向Subscriber发送错误事件，这个方法只能被调用1次。
• onComplete:如果Observable已经发送完所有的数据，并且没有发生错误，这时需要 调用 这个方法来向Subscriber发送结束事件，这个方法也只能调用1次，而且和onError是互斥关系。

Subscriber提供了unsubscribe方法，当Subscriber订阅到Observable之后，可随时调用这个方法来终止对Observable的订阅。

除了通用的Observable之外，还有两种特殊的Observable，分别是Single和Completable，与之分别对应的Subscriber是SingleSubscriber和CompletableSubscriber。

2、操作符

2.1 创建操作符

 2.2 转换操作符

2.3 过滤操作符

2.4 组合操作符

 2.5 错误处理操作符

2.6 辅助操作符

2.7 条件操作

2.8 聚合操作符

2.9 与Connectable Observable相关操作符

2.10 自定义操作符

需要遵守以下几条规则

• 自定义操作符在发送任何数据之前都要使用 subscriber.isUnsubscribed()来检查Subscriber的状态，如果没有任何Subscriber订阅就没有必要去发送数据了。
• 自定义操作要遵循Observable的核心原则

1. 可以多次调用Subscriber的onNext方法，但是同一个数据只能调用一次。
2. 可以调用Subscriber的onComplete或者onError方法，但这两个方法是互斥的。
3. 如果无法保证遵守以上两条原则，可以对自定义操作符加上serialize操作符，这个操作符会强制发送正常的数据。

• 自定义操作符内部不能阻塞住
• 如果通过组合多个RxJava原生的操作符就能达到目的，就不要使用自定义操作符来实现。

3、Scheduler

3.1 类型

有computation, io,newThread等类型。

computation Scheduler:适用于和CPU相关的任务，不适合那些会阻塞的任务。

newThread Scheduler:每次都会新建一个线程，一般情况下不推荐使用这个，因为创建一个线程都会造成稍微的延迟。

io Scheduler:类似于newThread Scheduler,io Scheduler可以被回收利用。内部会维持一个线程池，当使用io Scheduler来处理任务的时候 ，会首先从线程池中查找空闲的线程，如果有空闲线程就会在这个空闲线程上执行任务，否则会创建一个新的线程来执行任务，并在任务执行完毕时将这个空闲的线程放入到线程池中。当然空闲的线程不会一直在那里等待，RxJava默认空闲线程的存活时间是60秒，空闲时间超过60秒的线程会被回收。适合那些使用很少CPU资源的I/O操作。

immediate Scheduler：会在当前的线程上立即开始执行任务，这会将当前线程正在进行的任务阻塞。

trampoline Scheduler:与immediate Scheduler相似，在当前线程上执行任务，不是立即开始执行任务的，而是等待当前线程上之前的任务都结束之后再开始执行。

3.2 类抽象

3.3  类层次图

4、RxJava实现原理

4.1 订阅

调用Observable的subscribe方法，会调用另一个静态subscribe方法将Observable自身作为一个参数传入。

1.  
   static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
2.  
   // validate and proceed
3.  
   if (subscriber == null) {
4.  
   throw new IllegalArgumentException("subscriber can not be null");
5.  
   }
6.  
   if (observable.onSubscribe == null) {
7.  
   throw new IllegalStateException("onSubscribe function can not be null.");
8.  
   /*
9.  
   * the subscribe function can also be overridden but generally that's not the appropriate approach
10.  
    * so I won't mention that in the exception
11.  
    */
12.  
    }
13.  
     
14.  
    // new Subscriber so onStart it
15.  
    subscriber.onStart();
16.  
     
17.  
    /*
18.  
    * See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
19.  
    * to user code from within an Observer"
20.  
    */
21.  
    // if not already wrapped
22.  
    if (!(subscriber instanceof SafeSubscriber)) {
23.  
    // assign to `observer` so we return the protected version
24.  
    subscriber = new SafeSubscriber<T>(subscriber);
25.  
    }
26.  
     
27.  
    // The code below is exactly the same an unsafeSubscribe but not used because it would
28.  
    // add a significant depth to already huge call stacks.
29.  
    try {
30.  
    // allow the hook to intercept and/or decorate
31.  
    RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
32.  
    return RxJavaHooks.onObservableReturn(subscriber);
33.  
    } catch (Throwable e) {
34.  
    // special handling for certain Throwable/Error/Exception types
35.  
    Exceptions.throwIfFatal(e);
36.  
    // in case the subscriber can't listen to exceptions anymore
37.  
    if (subscriber.isUnsubscribed()) {
38.  
    RxJavaHooks.onError(RxJavaHooks.onObservableError(e));
39.  
    } else {
40.  
    // if an unhandled error occurs executing the onSubscribe we will propagate it
41.  
    try {
42.  
    subscriber.onError(RxJavaHooks.onObservableError(e));
43.  
    } catch (Throwable e2) {
44.  
    Exceptions.throwIfFatal(e2);
45.  
    // if this happens it means the onError itself failed (perhaps an invalid function implementation)
46.  
    // so we are unable to propagate the error correctly and will just throw
47.  
    RuntimeException r = new OnErrorFailedException("Error occurred attempting to subscribe [" e.getMessage() "] and then again while trying to pass to onError.", e2);
48.  
    // TODO could the hook be the cause of the error in the on error handling.
49.  
    RxJavaHooks.onObservableError(r);
50.  
    // TODO why aren't we throwing the hook's return value.
51.  
    throw r; // NOPMD
52.  
    }
53.  
    }
54.  
    return Subscriptions.unsubscribed();
55.  
    }
56.  
    }

订阅时序图为

订阅的过程就是调用 Observable内部的成员变量onSubscribe的call方法，将Subscriber作为参数传入进去，然后由onSubscribe对象向Subscriber发送数据的过程。在不使用Scheduler时，Observable和Subscriber都会运行在订阅时的线程上。

4.2 操作符实现

4.2.1 lift工作原理

 OnSubscribeLift类结构

 创建Observable时序

lift订阅时的时序

 创建一个liftObservable，liftObservable的onSubscribe成员变量是一个OnSubscribeLift对象，该对象持有一个Observable的onSubscribe成员变量的引用。当有Subscriber订阅的时候会调用这个OnSubscribeLift对象的call方法，调用Operator的call方法返回一个包装后的Subscriber对象，然后再调用源onSubscribe的call方法将liftSubscriber 传进去。源Observable在发送数据时，会先发送到liftSubscriber，由liftSubscriber对数据做一些自定义的操作，然后再发送给原始的Subscriber.

4.2.2 map工作原理 

类结构为：

依赖于OnSubscribeMap类，并且会将当前的Observable对象和Func1对象一起作为构造函数，然后创建并返回一个新的Observable，称之为MapObservable。

订阅时序图为

 订阅时调用call方法会做如下操作

• 创建一个新的MapSubscriber:parent
• 将parent通过add方法添加到源Subscriber中，这样在取消订阅时能够将所有Subscriber都取消
• 将parent订阅到源Observable中。

4.2.3 flatMap工作原理 

首先判断一下当前的Observable是否是ScalarSynchronousObservable。普通的Observable是先map然后再merge。

通过map结合传入的Func1对象将源Observable发送的数据转化为一个个小的Observable，然后通过merge将所有小的Observable组合成一个Observable后返回 。

OperateMerge类结构 

OperatorMerge操作符作用时创建MergeSubscriber,包含了源Subscriber。

 当MergeSubscribe接收到数据时，创建InnerSubscriber，同时让接收到的数据Observable订阅InnserSubscriber。同时启动发送数据emit。将数据发送给源Subscriber

 InnserSubscriber接收到数据时，会调用MergeSubscriber的tryEmit将数据发送给源Subscriber

其处理图为

4.2.4 concat工作原理

其方法为

 

 

其类结构图为

订阅时序图为

 

5、Subject

 PublishSubject:接收订阅之后的所有数据

ReplaySubject:接收订阅前后的所有数据

BehaviorSubject:接收订阅前的最近一个及订阅后的所有数据

AsyncSubject:只接收最后一条数据

6、BlockingObservable

 转换成Future通过使用BlockingOperatorToFuture操作符

1.  
   public static <T> Future<T> toFuture(Observable<? extends T> that) {
2.  
    
3.  
   final CountDownLatch finished = new CountDownLatch(1);
4.  
   final AtomicReference<T> value = new AtomicReference<T>();
5.  
   final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
6.  
    
7.  
   @SuppressWarnings("unchecked")
8.  
   final Subscription s = ((Observable<T>)that).single().subscribe(new Subscriber<T>() {
9.  
    
10.  
    @Override
11.  
    public void onCompleted() {
12.  
    finished.countDown();
13.  
    }
14.  
     
15.  
    @Override
16.  
    public void onError(Throwable e) {
17.  
    error.compareAndSet(null, e);
18.  
    finished.countDown();
19.  
    }
20.  
     
21.  
    @Override
22.  
    public void onNext(T v) {
23.  
    // "single" guarantees there is only one "onNext"
24.  
    value.set(v);
25.  
    }
26.  
    });
27.  
     
28.  
    return new Future<T>() {
29.  
     
30.  
    private volatile boolean cancelled;
31.  
     
32.  
    @Override
33.  
    public boolean cancel(boolean mayInterruptIfRunning) {
34.  
    if (finished.getCount() > 0) {
35.  
    cancelled = true;
36.  
    s.unsubscribe();
37.  
    // release the latch (a race condition may have already released it by now)
38.  
    finished.countDown();
39.  
    return true;
40.  
    } else {
41.  
    // can't cancel
42.  
    return false;
43.  
    }
44.  
    }
45.  
     
46.  
    @Override
47.  
    public boolean isCancelled() {
48.  
    return cancelled;
49.  
    }
50.  
     
51.  
    @Override
52.  
    public boolean isDone() {
53.  
    return finished.getCount() == 0;
54.  
    }
55.  
     
56.  
    @Override
57.  
    public T get() throws InterruptedException, ExecutionException {
58.  
    finished.await();
59.  
    return getValue();
60.  
    }
61.  
     
62.  
    @Override
63.  
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
64.  
    if (finished.await(timeout, unit)) {
65.  
    return getValue();
66.  
    } else {
67.  
    throw new TimeoutException("Timed out after " unit.toMillis(timeout) "ms waiting for underlying Observable.");
68.  
    }
69.  
    }
70.  
     
71.  
    private T getValue() throws ExecutionException {
72.  
    final Throwable throwable = error.get();
73.  
     
74.  
    if (throwable != null) {
75.  
    throw new ExecutionException("Observable onError", throwable);
76.  
    } else if (cancelled) {
77.  
    // Contract of Future.get() requires us to throw this:
78.  
    throw new CancellationException("Subscription unsubscribed");
79.  
    } else {
80.  
    return value.get();
81.  
    }
82.  
    }
83.  
     
84.  
    };
85.  
     
86.  
    }

使用CountDownLatch来实现，源Observable订阅特定的Subscriber，在处理onNext时，设置返回结果，在处理onComplete时，countDownLatch减1，在处理onError时设置error值，同时countDownLatch减1结束。在future中获取值时，等待countDownLatch

参考资料：

https://github.com/ReactiveX/RxJava

这篇好文章是转载于：学新通技术网