Rxjava入门指南(一)：响应式编程与Rxjava入门

迷南。 2023-10-11 10:07 45阅读 0赞

## 响应式编程介绍 ##

先来介绍一下响应式编程：响应式编程是一种，基于事件驱动的方式，处理异步数据(事件)流的编程范式。

实际上就是 观察者模式+数据流 + 事件控制

**1.什么算是数据流？**

举个例子，在界面中点击登录按钮发出登录请求，这个事件就是一个数据流。

**2.什么算是基于事件驱动？**

打个比方，发出登录请求相当于拧开水龙头(水龙头发生变化)，而执行登录请求就相当于接收水的水池(水池发生变化)，如果水龙头有水流出时，水池可以接收水管中的水。

![1696704-20190822150128958-664890967.png][]

水池可以根据水龙头发出的请求而产生响应，即从对事件的发生和对事件的响应这个角度来编程。

**3.怎样算是响应？**

举个维基百科中的例子：如果我们声明变量a = b + c，那么当b的值改变后，a的值是不会发生改变的，而响应式编程则是当b发生改变后，可以让a**响应**从而发生变化。

和观察者模式一样，当一个对象改变状态，会通知其他订阅者。

**4.那为什么是异步的？**因为拧开水龙头和水池接收水不是线性操作：

假如有水龙头A和B，水池正在接收水龙头A中的水，当水龙头B中的水流向水池时，不会**阻塞**水池接收水龙头A中的水，两者是不相关的。

同样，假如我们有水池A和水池B，当水龙头的水流向水池A的同时，不会**阻塞**水龙头的水流向水池B

所以说，事件的发出(拧开水龙头)和事件的接收(水池接收水)这两个动作是异步的，互不相扰的，不会像同步操作那样产生阻塞。

## Rxjava介绍(rxjava2) ##

rx全称为Reactive Extension，即响应式、可扩展，是一套基于事件的异步编程的API。

在Rxjava中，事件的发出对应：Observable(可观测的物)，事件的接收对应：Observer(观察者)，这个过程的连接对应着subscribe()方法(订阅)。

![1696704-20190822150518010-1122382138.png][]

我们来通过代码描述这个过程：

创建Observable对象(水龙头)

//创建可观测的物Observable：水龙头
            Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) {
                    //ObservableEmitter 事件发射器，可以发出的事件类型有：onNext、onError、onComplete
                    emitter.onNext("水流1"); //发出事件
                    emitter.onNext("水流2");
                    emitter.onNext("水流3");
                    emitter.onComplete(); //与onError互斥，只能发出两者中的一个
                    emitter.onNext("水流4"); // 当发出onComplete或onError后，虽然可以继续发出事件，但不会再被接收到
                }
            });

创建Observer对象(水池)

//创建观察者Observer：水池
            Observer<String> observer = new Observer<String>() {
                @Override
                public void onSubscribe(Disposable disposable) {
         
    　　　　　　　　　　//这里获取到了Disposable对象，下面会介绍该对象
                    Log.d(TAG,"subscribe:建立连接");
                }
    
                @Override
                public void onNext(String water) {
                    Log.d(TAG,"接收水流：" + water);
                }
    
                @Override
                public void onError(Throwable error) {
                    Log.d(TAG,"报错啦");
                }
    
                @Override
                public void onComplete() {
                   Log.d(TAG,"通知结束啦");
                }
            };

建立订阅关系(连接)

//建立连接：水管
            observable.subscribe(observer);

输出结果如下：

subscribe:建立连接
    接收水流：水流1
    接收水流：水流2
    接收水流：水流3
    通知结束啦

可以看到在建立连接后，依次接收到了水流1、2、3，而水流4没有被接收到。

我们可以把new Observer到Observable的subscirbe方法里，把整个过程连起来：

//创建可观测的物Observable：水龙头
            Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) {
                    //ObservableEmitter 事件发射器，可以发出的事件类型有：onNext、onError、onComplete
                    emitter.onNext("水流1"); //发出事件
                    emitter.onNext("水流2");
                    emitter.onNext("水流3");
                    emitter.onComplete(); //与onError互斥，只能发出两者中的一个
                    emitter.onNext("水流4"); // 当发出onComplete或onError后，会用Disposable对象的dispose()方法
                                            // 虽然水龙头可以继续发出事件，但水池不会再被接收到
    
                }
            }).subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable disposable) {
                    //这里可以获取到Disposable对象
                    Log.d(TAG,"subscribe:建立连接");
                }
    
                @Override
                public void onNext(String water) {
                    Log.d(TAG,"接收水流：" + water);
                }
    
                @Override
                public void onError(Throwable error) {
                    Log.d(TAG,"报错啦");
                }
    
                @Override
                public void onComplete() {
                    Log.d(TAG,"通知结束啦");
                }
            });

Observable的subscirbe方法还提供了其他几种重载形式：

public final Disposable subscribe() \{\}  
public final Disposable subscribe(Consumer<? super T> onNext) \{\}  
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) \{\}  
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) \{\}  
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) \{\}  
public final void subscribe(Observer<? super T> observer) \{\}

我们可以传入Consumer对象，让观察者处理自己关心的事件类型

Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) {
                    //ObservableEmitter 事件发射器，可以发出的事件类型有：onNext、onError、onComplete
                    emitter.onNext("水流1"); //发出事件
                    emitter.onNext("水流2");
                    emitter.onNext("水流3");
                    emitter.onComplete(); //与onError互斥，只能发出两者中的一个
                    emitter.onNext("水流4"); // 当发出onComplete或onError后，虽然可以继续发出事件，但不会再被接收到
                }
            }).subscribe(new Consumer<String>() {
                @Override
                public void accept(String str) {
                    Log.d(TAG, "onNext: " + str);
                }
            });

现在这个方法的返回结果不再是void而是返回了Disposable对象。

之前我们在Observer的onSubscribe中就提到了该对象，在onSubscribe方法中可以获取到Disposable对象，所以不需要返回对象。

## Disposable对象 ##

*  dispose():解除订阅
 *  isDisposed():查询是否解除了订阅

一般在activity或者fragment销毁时，我们可以调用dispose方法来取消订阅，释放数据流中剩余的所有内部资源，防止内存泄漏。

Rxjava提供了CompositeDisposable容器管理多个Disposable对象，我们可以在获取到Disposable对象后把它添加到容器中，然后在onDestroy方法中调用compositeDisposable.dispose()清除所有订阅：

//添加Disposable对象
        protected void addDisposable(Disposable disposable) {
            if (null == compositeDisposable || compositeDisposable.isDisposed()) {
                compositeDisposable = new CompositeDisposable();
            }
            compositeDisposable.add(disposable);
        }
    
        //清除所有订阅
        protected void clearDisposable() {
            compositeDisposable.clear();
        }

## 异步 ##

我们简单的实现了事件的发出和响应，但是这个过程还发生在同一个线程内，并不是异步的。

rxjava提供了方便的线程调度方法用于切换线程：

//事件的发出在新线程newThread()
    observable.subscribeOn(Schedulers.newThread())
    　　　　　　  //事件的响应在主线程                                              
                .observeOn(AndroidSchedulers.mainThread())
    　　　　　　　//建立连接                      
                .subscribe(consumer对象);

之前的例子可以就成了这样：

//创建可观测的物Observable：水龙头
            Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) {
                    //ObservableEmitter 事件发射器，可以发出的事件类型有：onNext、onError、onComplete
                    emitter.onNext("水流1"); //发出事件
                    emitter.onNext("水流2");
                    emitter.onNext("水流3");
                    emitter.onComplete(); //与onError互斥，只能发出两者中的一个
                    emitter.onNext("水流4"); // 当发出onComplete或onError后，虽然可以继续发出事件，但不会再被接收到
                }
            }).subscribeOn(Schedulers.newThread())
                    //事件的响应在主线程                                              
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<String>() { //建立连接
                        @Override
                        public void accept(String str) {
                            Log.d(TAG, "onNext: " + str);
                        }
                    });

这样就实现了异步，事件发出时所在的线程为RxNewThreadScheduler-1，而接收事件所在的线程为main。

RxJava内置的线程包括(这些线程之间的具体区别暂且不提，主要是讨论如何使用)：

*  AndroidSchedulers.mainThread() 主线程
 *  Schedulers.io() IO线程，线程名RxCachedThreadScheduler-1
 *  Schedulers.newThread() 新线程
 *  Schedulers.computation() 需要大量计算的线程

以后我们就可以在IO线程中执行读写文件、读写数据库、网络操作等操作，在computation线程中执行计算操作。

现在我们已经通过rxjava实现了简单的异步观察者模式，模拟出了事件响应。

## 应用举例： ##

Retrofit网络请求框架提供了对Rxjava的支持，在IO线程中进行网络请求，在主线程中去处理结果。

Disposable disposable = retrofit.create(ILoginApi.class).login(loginDTO)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<ResultVO<LoginVO>>() {
                                   @Override
                                   public void accept(ResultVO<LoginVO> loginResultVO) throws Exception {
                                       //成功获取
                                   }
                               },
                            new Consumer<Throwable>() {
                                @Override
                                public void accept(Throwable throwable) throws Exception {
                                    //打印异常信息
                                }
                            });

**参考：通俗易懂的rxjava2教程：https://www.jianshu.com/p/464fa025229e**

**如有纰漏，请大佬及时批评指出**

转载于:https://www.cnblogs.com/gss128/p/11394363.html

[1696704-20190822150128958-664890967.png]: https://img2018.cnblogs.com/blog/1696704/201908/1696704-20190822150128958-664890967.png
[1696704-20190822150518010-1122382138.png]: https://img2018.cnblogs.com/blog/1696704/201908/1696704-20190822150518010-1122382138.png