Android RxJava之基本使用（一）

前言

本系列主要是基于Carson_Ho所写的《RxJava系列教程》系列文章，对RxJava的相关知识点进行梳理，文中图片全部来自《RxJava系列教程》。这一些列文章对我帮助很大，通篇把相关知识点和生活上的例子相结合的讲解方式，一直是我本人比较欣赏的。特别是将RxJava的原理通过顾客到饭店吃饭这个生活例子来讲解，让我印象很深刻。因为学习的时候，很多概念性的东西都太过于抽象，使用联想法，将其和生活上的例子结合起来，增加趣味性之余，无疑降低了理解的难度，便于我们学习。对于我来说，学习新的东西，个人的一些心得体会就是，思考的顺序应该是这样的：
- RxJava是什么？
- RxJava用来干什么的？功能是什么？有哪些东西和它的动能是一样的吗？
- RxJava优势在哪里？为什么非要使用它？
- RxJava是怎么样使用的？
- RxJava的原理是什么？
- ........

定义

一个在Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。

作用

实现异步操作，类似于Android中的AsyncTask、Handler作用。

特点

基于事件流的链式调用，在程序逻辑异常复杂的情况下，仍然可以让代码的逻辑保持简洁。

原理

RxJava的异步实现，是通过一种扩展的观察者模式来实现的。被观察者（Observable）通过订阅（Subscribe）按顺序发送事件给观察者（Observer），观察者（Observer）按顺序接收事件和做出对应的响应动作。
- 被观察者(Observable)
  - 作用：决定什么时候触发事件以及触发什么样的事件。
  - 创建方法：
    - Observable.just(T...) 参数为单个。
    - Observable.from(T[])/Observable.fromArray(Iterable<? extends T>)参数为数组或Itera
- 观察者(Observice)
  - 作用：当事件触发的时候将有怎么样的行为。
  - 实现类有Observer/Subscriber
举个例子，顾客到饭店吃饭，如下图所示

基本使用

使用步骤
事件类型
示例代码

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "RxJava";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);


        //RxJava的基本使用
        Observable.create(new ObservableOnSubscribe<Integer>() {
            // 1. 创建被观察者 & 生产事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
                //观察者接收到Complete事件后将不再继续接收任何事件
                //所以事件4、事件5没有在日志中打印出来
                emitter.onNext(4); 
                emitter.onNext(5);

            }
        }).subscribe(new Observer<Integer>() {
            //2.通过订阅（subscribe）连接观察者和被观察者
            //3.创建观察者 & 定义响应事件的行为

            //默认最先调用复写的onSubscribe()方法
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件" + value + "作出响应");
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        });

        // 注：整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.onNext（）>观察者.onComplete()
    }
}

运行效果

案例
- 由指定的一个drawable文件id取得图片，并显示在ImageView中，并在出现异常的时候打印Toast报错。


    private void showImageDemo() {
        final int drawID = R.mipmap.ic_launcher;
        Observable
                .create(new ObservableOnSubscribe<Drawable>() {
                    @Override
                    public void subscribe(ObservableEmitter<Drawable> emitter) throws Exception {
                        Drawable drawable = ContextCompat.getDrawable(MainActivity.this, drawID);
                        emitter.onNext(drawable);
                        emitter.onComplete();

                    }
                })
                .subscribe(new Observer<Drawable>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Drawable drawable) {
                        imageView.setImageDrawable(drawable);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Toast.makeText(MainActivity.this, "Error", Toast.LENGTH_SHORT).show();
                    }
                    @Override
                    public void onComplete() {
                        
                    }
                });
    }

额外说明
- 观察者Observer的subscribe()具备多个重载的方法。
- 可采用Disposable.dispose()切断观察者与被观察者之间的连接。即观察者无法继续接收被观察者的事件，但被观察者还是可以继续发送事件。

        }).subscribe(new Observer<Integer>() {


            // 1. 定义Disposable类变量
            private Disposable mDisposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
                // 2. 对Disposable类变量赋值
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件" + value + "作出响应");
                if (value == 2) {
                    // 3. 设置在接收到第二个事件后切断了观察者和被观察者的连接。
                    mDisposable.dispose();
                    Log.d(TAG, "已经切断了连接" + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        });

运行效果