RxJava 源码解析篇

谁践踏了优雅 2022-02-24 14:40 261阅读 0赞

说实话rxjava 的确是我见过最绕，最强大，最复杂的类库之一，对于一般的开发者基本上都是望文却步！比如在[netflix hystrix][]的AbstractCommand中有使用RxJava，那简直是帅到了极致，从头到尾都是链式响应，非常高逼格。为了加深对它的理解，本文将对此展开对Observer、Observable、Subscribe的源码的探索。

//创建被观察者（也就是开关）：
    Observable.OnSubscribe<String> onSwitcherSubscribe = new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            System.out.println(Thread.currentThread().getName() + ",create");
            subscriber.onNext("On");
            subscriber.onNext("Off");
            subscriber.onCompleted();
        }
    };
    Observable switcherObservable = Observable.create(onSwitcherSubscribe);
    //创建观察者（也就是台灯）
    Subscriber lightSubscriber = new Subscriber<String>() {
        @Override
        public void onCompleted() {
            //被观察者的onCompleted()事件会走到这里;
            System.out.println(Thread.currentThread().getName() + " 结束观察...\n");
        }
        @Override
        public void onError(Throwable e) {
            //出现错误会调用这个方法
        }
        @Override
        public void onNext(String s) {
            //处理传过来的onNext事件
            System.out.println(Thread.currentThread().getName() + ",onNext:" + s);
        }
    };
    //创建订阅
    switcherObservable.subscribe(lightSubscriber);

**运作的流程**

![20190404141840325.gif][]

也就是说subscriber（）很重要，它是整个联调运作的激活点！

那它究竟做了什么呢？

1.  调用 `Subscriber.onStart()` ，`onStart`是一个可选的准备方法
2.  调用 `Observable` 中的 `OnSubscribe.call(Subscriber)`，事件发送的逻辑开始运行。从这也可以看出，在 RxJava 中， `Observable` 并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候，即当 `subscribe()` 方法执行的时候
3.  将传入的 `Subscriber` 作为 `Subscription` 返回，返回为了方便 `unsubscribe()`

如何验证呢？

看源码，源码比较复杂，其实做的事情不复杂（简述）：

> public Subscription subscribe(Subscriber lightSubscriber) \{  
>     lightSubscriber.onStart();  
>     onSwitcherSubscribe.call(lightSubscriber);  
>     return lightSubscriber;  
> \}

public class Observable<T> {
        //就是onSwitcherSubscribe
        final Observable.OnSubscribe<T> onSubscribe;
        public final Subscription subscribe(Subscriber<? super T> subscriber) {
            return subscribe(subscriber, this);
        }
        private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
            if (subscriber == null) {
                throw new IllegalArgumentException("observer can not be null");
            } else if (observable.onSubscribe == null) {
                throw new IllegalStateException("onSubscribe function can not be null.");
            } else {
                //先调用了lightSubscriber.onStart
                ((Subscriber)subscriber).onStart();
                if (!(subscriber instanceof SafeSubscriber)) {
                    subscriber = new SafeSubscriber((Subscriber)subscriber);
                }
                try {
                    //最终调用onSwitcherSubscribe.call（参数：lightSubscriber）
                    hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
                    //返回lightSubscriber
                    return hook.onSubscribeReturn((Subscription)subscriber);
                } catch (Throwable var6) {
                    Throwable e = var6;
                    Exceptions.throwIfFatal(var6);
    
                    try {
                        ((Subscriber)subscriber).onError(hook.onSubscribeError(e));
                    } catch (Throwable var5) {
                        Exceptions.throwIfFatal(var5);
                        RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + var6.getMessage() + "] and then again while trying to pass to onError.", var5);
                        hook.onSubscribeError(r);
                        throw r;
                    }
    
                    return Subscriptions.unsubscribed();
                }
            }
        }
    }

那比如：Observable.just("a", "b").subscribe(x -> System.out.println(x));可以理解吗？

直接看源码，不需要再解释了吧

public class Observable<T> {
        public static <T> Observable<T> just(T t1, T t2) {
            return from((Object[])(new Object[]{t1, t2}));
        }
        public static <T> Observable<T> from(T[] array) {
            int n = array.length;
            if (n == 0) {
                return empty();
            } else {
                return n == 1 ? just(array[0]) : create(new OnSubscribeFromArray(array));
            }
        }
        public static <T> Observable<T> create(Observable.OnSubscribe<T> f) {
            return new Observable(hook.onCreate(f));
        }
    }
    public final class OnSubscribeFromArray<T> implements OnSubscribe<T> {
        final T[] array;
        public OnSubscribeFromArray(T[] array) {
            this.array = array;
        }
    }

这时，你可能会问`Observer` 和 `Subscriber的区别？看之前的文章吧！`![89032294][]

[netflix hystrix]: https://blog.csdn.net/alex_xfboy/article/details/88720949
[20190404141840325.gif]: /images/20220224/098303f7b9fa49ff8cea877e76417c7a.png
[89032294]: /images/20220224/bd5648ef717349ab8dab667cb633785a.png