Rxjava 的源码解析 (三) RxJava2 基本元素 (无背压)

刺骨的言语ヽ痛彻心扉 2022-05-23 23:04 178阅读 0赞

### 文章目录 ###

*   *  涉及到得其他知识
     *   *  AtomicReference
     *  使用版本
     *  先看一下简单案例
     *  基本元素
     *   *  Observable
         *  Observer
         *  Disposable
         *  ObservableOnSubscribe
         *  Emitter
     *  简单案例的源码分析
     *  流程思维图

> 其实设计上 基本与 RxJava1 ，优化了一些东西

## 涉及到得其他知识 ##

### AtomicReference ###

https://blog.csdn.net/chuchus/article/details/50801993

## 使用版本 ##

compile "io.reactivex.rxjava2:rxjava:2.0.0"

## 先看一下简单案例 ##

Observable.create(new ObservableOnSubscribe<String>() {  // tag1
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception { 
            if (e.isDisposed()){ 
                e.onNext("发送消息");
                e.onComplete();
            }
        }
    }).subscribe(new Observer<String>() {   //tag2
        @Override
        public void onSubscribe(Disposable d) { 
            System.out.println("onSubscribe");
        }
    
        @Override
        public void onNext(String value) { 
            System.out.println("onNext -> " + value);
        }
    
        @Override
        public void onError(Throwable e) { 
            System.out.println("onError");
        }
    
        @Override
        public void onComplete() { 
            System.out.println("onComplete");
        }
    });

## 基本元素 ##

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

### Observable ###

> 与 RxJava1 没有明显区别

*  观察得到的 ----- 被观察者，不支持背压
 *  通过Observable创建一个可观察的序列（create方法）
 *  通过subscribe注册观察者

### Observer ###

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

### Disposable ###

*  与 RxJava1 的subscription的作用相当
 *  用于取消订阅和获取当前订阅状态

### ObservableOnSubscribe ###

> 与 RxJava中 的 onSubscribe 相当

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

### Emitter ###

*  一个发射数据的接口，和 Observable的方法类似
 *  本质是对 Observer 和 Subscriber的包装

## 简单案例的源码分析 ##

> 我们这里暂时都不管 RxJava的 Hook，使用去除无用我们理解的语句解析

// tag1.create方法
    Observable.create(new ObservableOnSubscribe<String>(){ ...})
    
    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) { 
       // 其实内部相当于 使用 ObservableCreate 包裹后，返回 ObservableCreate 对象
      return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

ObservableCreate 继承于 Observable

// tag2
    public final void subscribe(Observer<? super T> observer) { 
       try { 
           ..
           // 这里相当于实际调用了，ObservableCreate 的 subscribeActual 并传入 observer
           // tag3
           subscribeActual(observer);
       } catch (NullPointerException e) { 
           throw e;
       } catch (Throwable e) { 
          ....
       }
    }

ObservableCreate 做了什么

public final class ObservableCreate<T> extends Observable<T> { 
        final ObservableOnSubscribe<T> source;
    
        public ObservableCreate(ObservableOnSubscribe<T> source) { 
            this.source = source;
        }
    
    	// tag3 将调用这个方法
        @Override
        protected void subscribeActual(Observer<? super T> observer) { 
            CreateEmitter<T> parent = new CreateEmitter<T>(observer);
            observer.onSubscribe(parent);
            try { 
                source.subscribe(parent);
            } catch (Throwable ex) { 
                Exceptions.throwIfFatal(ex);
                parent.onError(ex);
            }
        }
    }

ObservableCreate.CreateEmitter 内部类

static final class CreateEmitter<T>
        extends AtomicReference<Disposable>
        implements ObservableEmitter<T>, Disposable {
    
       // 这个对象也就是 Observable对象.subscribe 方法
       // 创建的 Observer
       final Observer<? super T> observer;
    
       CreateEmitter(Observer<? super T> observer) {
           this.observer = observer;
       }
    
       @Override
       public void onNext(T t) {
           if (t == null) {
               onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
               return;
           }
           if (!isDisposed()) {
               observer.onNext(t);
           }
       }
    	....
    
    }

## 流程思维图 ##

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

[70]: /images/20220524/add34c1aedf747fe92e30f6699c25081.png