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 详解

巴特西

RxJava 1.x 理解-1

先看下别人实现的最基本的RxJava的实现方式：

Subscriber（观察者）

Observable（订阅源）

实践

对比下Android的OnClickListener的实现

最新文章

热门文章