Rxjava 的源码解析 (二) RxJava1 基本元素分析

快来打我* 2022-05-25 14:16 150阅读 0赞

### 文章目录 ###

*   *  介绍
     *  讲解使用的库
     *   *  简单使用
         *  五大主要元素
         *   *  Observable
             *  Observer
             *  Subscription
             *  OnSubscribe
             *  Subscriber
         *  结合解释
         *  代码流程图

## 介绍 ##

> 部分资料来源于慕客网，加入个人自己的理解

## 讲解使用的库 ##

compile 'io.reactivex:rxjava:1.3.0'

### 简单使用 ###

Subscription subscription = Observable.create(new Observable.OnSubscribe<String>() // 关键 tag 1 {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            if (!subscriber.isUnsubscribed()) {
                subscriber.onNext("1");
                subscriber.onNext("2");
                subscriber.onCompleted();
            }
        }
    }).subscribe(new Observer<String>() {   // 关键tag2
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
        }
    
        @Override
        public void onError(Throwable e) {
            e.printStackTrace();
            System.out.println("onError");
    
        }
    
        @Override
        public void onNext(String s) {
            System.out.println("onNext：" + s);
        }
    });
    // 输出结果
    onNext：1
    onNext：2
    onCompleted

### 五大主要元素 ###

![这里写图片描述][70]

#### Observable ####

*  观察得到的 — 被观察者
 *  通过 Observable 创建一个可观察的序列 （create方法）
 *  通过 subscribe 去注册一个观察者

#### Observer ####

*  用于接收数据 — 观察者
 *  作为 Observable 的 subscribe方法的参数

#### Subscription ####

*  订阅，用于描述被观察者和观察者之间的关系
 *  用于取消订阅和获取当前的订阅状态

#### OnSubscribe ####

*  当订阅是会触发此接口调用
 *  在Observable内部，实际作用是向订阅者发射数据

#### Subscriber ####

*  实现了 Observer 和 Subscription
 *  只有自己才能阻止自己

### 结合解释 ###

关键 tag 1

// 上面代码创建对象
    Observable.create(new Observable.OnSubscribe<String>....)
    // 1. 实际将 创建的 Observable.OnSubscribe 赋值给 Observable 的成员变量
    Observable.this.onSubscribe = f

关键 tag 2

// 10220 行 
    public final Subscription subscribe(final Observer<? super T> observer) {
        if (observer instanceof Subscriber) {
            return subscribe((Subscriber<? super T>)observer);
        }
        if (observer == null) {
            throw new NullPointerException("observer is null");
        }
        // 1. 将传入进来的 Observer 包装为 ObserverSubscriber 
        // 2. 调用 subscribe
        return subscribe(new ObserverSubscriber<T>(observer));
    }
    
    // 10318 行 
    public final Subscription subscribe(Subscriber<? super T> subscriber) {
        return Observable.subscribe(subscriber, this);
    }
    
    // 10322行
     static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
     
            // .....
            subscriber.onStart();
    
            try {
                // 相当于调用了 Observable.this.onSubscribe 的 call方法
                RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
                return RxJavaHooks.onObservableReturn(subscriber);  // 内部什么都没做，直接 return subscriber
            } catch (Throwable e) {
    	        ....
                return Subscriptions.unsubscribed();
            }
        }

也就是调用了

![这里写图片描述][70 1]

### 代码流程图 ###

https://www.processon.com/view/link/5afaacb8e4b0ee8894b7e15b

[70]: /images/20220525/30b7c6a374124be8a17aa44cc59b6821.png
[70 1]: /images/20220525/e97b6d2f5c1c4ff8b73883f326d870b7.png