RxJava 1.x 理解-1

先看下別人實現的最基本的RxJava的實現方式：

在RxJava裏面，有兩個必不可少的角色：Subscriber（觀察者） 和 Observable（訂閱源）。

Subscriber（觀察者）

Subsribler在RxJava裏面是一個抽象類，它實現了Observer接口。

public interface Observer<T> {
    void onCompleted();
    void onError(Throwable t);
    void onNext(T var1);
}

爲了儘量的簡單，將Subscriber簡化以下：

public abstract class Subscriber<T> implements Observer<T> {
    public void onStart() {
    }
}

Observable（訂閱源）

Observable（訂閱源）在RxJava裏面是一個大而雜的類，擁有不少工廠方法和各式各樣的操做符。每一個Observable裏面有一個OnSubscribe對象，只有一個方法（void call(Subscriber<? super T> subscriber);），用來產生數據流，這是典型的命令模式

public class Observable<T> {
    final OnSubscribe<T> onSubscribe;

    private Observable(OnSubscribe<T> onSubscribe) {
        this.onSubscribe = onSubscribe;
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
        return new Observable<T>(onSubscribe);
    }

    public void subscribe(Subscriber<? super T> subscriber) {
        subscriber.onStart();
        onSubscribe.call(subscriber);
    }

    public interface OnSubscribe<T> {
        void call(Subscriber<? super T> subscriber);
    }
}

實踐

到此，一個小型的RxJava的雛形就出來了。不信？咱們來實踐一下吧。

 Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 10; i++) {
                    subscriber.onNext(i);
                }
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }
            @Override
            public void onError(Throwable t) {

            }
            @Override
            public void onNext(Integer var1) {
                System.out.println(var1);
            }
        });

OK，先來消化下最基本的知識：

首先Observable（訂閱源）內部有個接口 OnSubscribe ，其中有個 call 方法。

其次Observable（訂閱源）內部持有  final OnSubscribe<T> onSubscribe; 屬性，但沒有實例化，它在什麼地方實例化呢，在 create 方法中，實例下這個 onSubscribe 屬性，而且須要實現 call 方法。

接着Observable（訂閱源）執行 subscribe 觸發事件，其中  onSubscribe.call(subscriber); 則回調執行實例化的 onSubscribe 且實現的 call 方法。

對比下Android的OnClickListener的實現

    Button.setOnClickListener(new OnClickListener(){  
        @Override  
        public void onClick(View v) {  
           // do something      
        }         
    });  

Button至關於Observable；---> 被監聽的對象

OnClickListener至關於OnSubscribe；---> 接口

onClick至關於call；---> 接口回調方法

當咱們點擊按鈕的時候，點擊的事件系統分發到Button上，相似subscribe，回調執行onClick方法，一般狀況下，咱們對onClick（View v）中的View不感興趣，但其實，這個View就是Button，咱們能夠對其進行操做。

        button = (Button) findViewById(R.id.button);
        button.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                Log.d(TAG, "view instanceof Button:" + (view instanceof Button));
                Log.d(TAG, "view == button:" + (view == button));
                view.setVisibility(View.GONE);
            }
        });

因此在使用上能夠簡單的理解爲：

輸入的數據 ---> 被監聽者（訂閱源）對這些數據進行操做，或者執行響應的處理 --> 產生新的數據，或者事件發送給監聽者 --> 監聽者執行本身的方法。

        Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("hello rxjava");
            }
        }).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                Log.d(TAG, s);
            }
        });

 

參考資料：

一塊兒來造一個RxJava，揭祕RxJava的實現原理

給 Android 開發者的 RxJava 詳解