Android-RxJava源码分析

深藏阁楼爱情的钟 2022-05-24 22:22 117阅读 0赞

# 最简单的源码分析 #

//创建一个上游 Observable：
    Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(3);
            emitter.onComplete();
        }
    });
    //创建一个下游 Observer
    Observer<Integer> observer = new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "subscribe");
        }
    
        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "" + value);
        }
    
        @Override
        public     void onError(Throwable e) {
            Log.d(TAG, "error");
        }
    
        @Override
        public void onComplete() {
            Log.d(TAG, "complete");
        }
    };
    //建立连接
    observable.subscribe(observer);

## 事件流 ##

![70][]

## UML类图 ##

![70 1][]

## 函数调用流程图 ##

![70 2][]

# Map操作符源码分析 #

//上游操作1
            Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    emitter.onNext(1);
                    emitter.onNext(2);
                    emitter.onNext(3);
                }
            });
            //上游操作2
            Observable<String> mapObservable = observable.map(new Function<Integer, String>() {
                @Override
                public String apply(Integer integer) throws Exception {
                    return "This is result " + integer;
                }
            });
            //下游操作
            Observer<String> observer = new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "subscribe");
                }
    
                @Override
                public void onNext(String value) {
                    Log.d(TAG, "" + value);
                }
    
                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, "error");
                }
    
                @Override
                public void onComplete() {
                    Log.d(TAG, "complete");
                }
            };
            //订阅
            mapObservable.subscribe(observer);

## 事件流 ##

![70 3][]

## UML类图 ##

![70 4][]

## 函数调用流程图 ##

![70 5][]

# 线程切换源码分析 #

//上游操作
            Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
                    Log.d(TAG, "emit 1");
                    emitter.onNext(1);
                }
            });
            //下游操作
            Observer<Integer> observer = new Observer<Integer>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "subscribe");
                }
    
                @Override
                public void onNext(Integer value) {
                    Log.d(TAG, "Observer thread is : " + Thread.currentThread().getName());
                    Log.d(TAG, "" + value);
                }
    
                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, "error");
                }
    
                @Override
                public void onComplete() {
                    Log.d(TAG, "complete");
                }
            };
            //订阅操作
            observable
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(observer);

**subscribeOn的UML类图**

![70 6][]

## observeOn的UML类图 ##

![70 7][]

## 事件流及函数调用流程图 ##

事件流和函数调用流程与Map操作符类似，通过责任链模式，将切换线程操作包装成一个管道，然后在管道中进行Observer和Observable的线程切换操作。

[70]: /images/20220525/46d359290385445cbca5d281d2d0b4a1.png
[70 1]: /images/20220525/ad3114a0bc66406fab9c6f9e4ffae597.png
[70 2]: /images/20220525/493ac8650c8f4406a54da578fcbb16a7.png
[70 3]: /images/20220525/93d3e02c011e4fbebd89957a2014e7aa.png
[70 4]: /images/20220525/8c22690738014a8cb103761464391bb1.png
[70 5]: /images/20220525/1c4fac29d3da4bc5920f649e364f672c.png
[70 6]: /images/20220525/9040b3377f9e4381b2349d49a2ddd431.png
[70 7]: /images/20220525/e65a290de9214bcca7241daa610b0b0e.png