image.png

背景

响应式编程

响应式编程简单点来说可以称为异步数据流编程，当一段逻辑关系变更之后，可以及时响应，并处理变更后的逻辑关系，我们称之为响应式编程。下面我们通过一个示例来看下。

int a=1;
int b=a+1; 
System.out.print(“b=”+b)    //  b=2 
a=10; System.out.print(“b=”+b)    //  b=2

我们希望达到，不管a的值如何变化，b的值永远比a大1，如果仅靠赋值更新的话，我们需要另外花精力去维护a和b的关系。而响应式编程则是希望通过某种操作符来构建这种关系：

int a=1;
int b <= a+1;   // <= 符号只是表示a和b之间关系的操作符 
System.out.print(“b=”+b)    //  b=2 
a=10; System.out.print(“b=”+b)    //  b=11

这就是是响应式的思想，它希望有某种方式能够构建关系，而不是执行某种赋值命令。

Rx是响应式拓展，即支持响应式编程的一种拓展,为响应式在不同语言中的实现提供指导思想

如何理解RxJava

上面我们说到Rx是响应式编程的拓展，为了理解RxJava，下面我们通过一个日常实现的场景是描述。开发一个App，必然会涉及到App初始化，数据库初始化和登录等业务操作，登录完成之后，再跳转到新的页面。

场景示例图：

image.png

为了实现上面的业务逻辑，我们传统的做法可能是：

// 业务1 
initDB(context) 
// 业务2 loginApp(getUserData(),new OnLoginCallback{  
     // 业务3  
     void onLoginSucc(){ 
           startActivity()  // 业务4
     }
 })

上面这段代码，是比较常见的实现方式，但是业务之间是严格按照顺序执行，并不能反映实际的业务之间关系，它们之间真实的业务关系应该是分别完成业务1，2，3，然后再执行业务4。至此，为了构建业务之间的真实关系，RxJava此时要派上用场了。

Observable.just(context)  
                  .map((context)->{loginApp(getUserData())})  // 执行业务3 
                  .map((context)->{initApp(context)}) // 执行业务1 
                  .map((context)->{initDB(context)}) // 执行业务2 
                  .subscribeOn(Schedulers.newThread()) 
                  .subscribe((context)->{startActivity()})  // 执行业务4

很多同学可能会直接用上面这段代码来实现和构建上面的业务关系，但是上面实现并不是响应式的，本质上还是在同一个线程来实现，依旧是按顺序来执行，没有真实反应业务之间的关系，那么我们应该怎么去做呢？

// 初始化App 
Observable obserInitSDK = Observable.create((context)->{   
                                          initApp(context)}).subscribeOn(Schedulers.newThread())
 // 初始化数据库 
Observable obserInitDB = Observable.create((context)->{ 
                                       initDB(context)}).subscribeOn(Schedulers.newThread())
// 完成登陆 
Observable obserLogin = Observable.create((context)->{
                                       loginApp(getUserData())})  .map((isLogin)->{returnContext()}) .subscribeOn(Schedulers.newThread()) 
// 分别完成三个业务
Observable observable = Observable.merge(obserInitSDK,obserInitDB,obserLogin) 
//分开完成三个业务之后跳转到下一个 
observable.subscribe(()->{startActivity()})

大家应该能很明显看到两段代码的区别，下面这段代码完全遵照了业务之间客观存在的关系，可以说代码和业务关系是完全对应的。不同的业务分别是在不同线程异步执行，当业务中存在比较耗时的操作时，响应式代码可以极大的提高程序的执行效率，降低线程阻塞。

RxJava使用

RxJava使用通常分为三步：

创建一个被观察者(Observable)

被观察者类似一个任务栈，在给定序列的情况下，会循环遍历栈内的任务，尽管在订阅者在订阅它们之前不会发送数据，但是每执行一个操作，被观察者都会通过 onNext（）方法通知其指定的观察者，一旦执行完或者中途异常，都会通过下面两个方法来终止。

onComplete()：执行完成时调用
onError()：抛出异常时调用

private Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() { 
    @Override public void subscribe(ObservableEmitter<String> emitter) throws Exception { 
                              Log.e(TAG,"subscribe()"); emitter.onNext("next()-->1"); emitter.onNext("next()-->2"); 
                              emitter.onComplete(); 
                      }
  });

创建一个观察者（Observer）

我们可以简单的将观察者理解成对被观察者动作反应的角色，当观察者订阅了一个事件之后，被观察者发送或者传输数据时，观察者都会对齐做出反应。

比如上面被观察者在发送了两次onNext()事件后，通过下面的执行结果，我们可以看到观察者中的onNext()分别回应了被观察者发出的两次onNext()操作。

Observer<String> observer = new Observer<String>() { 
 @Override public void onSubscribe(Disposable d) {
        Log.i(TAG, "onSubscribe"); 
  }  
 @Override public void onNext(String s) { 
        Log.i(TAG, "onNext:" + s);
 }   
@Override public void onError(Throwable e) { 
        Log.i(TAG, "onError"); 
}  
@Override public void onComplete() { 
        Log.i(TAG, "onComplete");
 }
 };

创建订阅关系

observable.subscribe(observer);

执行结果

2020-09-04 16:33:46.502 30049-30049/com.qtopay.rxjava20 I/MainActivity-Observer: onSubscribe
2020-09-04 16:33:46.502 30049-30049/com.qtopay.rxjava20 E/MainActivity-Observer: subscribe()
2020-09-04 16:33:46.502 30049-30049/com.qtopay.rxjava20 I/MainActivity-Observer: onNext:next()-->1 
2020-09-04 16:33:46.502 30049-30049/com.qtopay.rxjava20 I/MainActivity-Observer: onNext:next()-->2 
2020-09-04 16:33:46.502 30049-30049/com.qtopay.rxjava20 I/MainActivity-Observer: onComplete

操作符简介

just()

Observable.just（）

你可以使用.just（）运算符将任何对象转换为Observable。结果Observable将发出原始对象并完成。

例如，这里我们创建一个Observable，它将向其所有Observers发出一个字符串：

Observable <String> observable = Observable.just（“Hello World！”）;

fron（）

Observable.from（）

.from（）运算符允许你将对象集合转换为可观察的流。你可以使用Observable.fromArray将数组转换为Observable，使用Observable.fromCallable将Callable转换为Observable，使用Observable.fromIterable将Iterable转换为Observable。

range()

Observable.range（）

你可以使用.range（）运算符发出一系列连续的整数。你提供的第一个整数是初始值，第二个是你要发出的整数。例如：

Observable <Integer> observable = Observable.range（0,5）;

interval()

Observable.interval（）

此运算符创建一个Observable，它发出无限的上升整数序列，每个发射由你选择的时间间隔分隔。例如：

Observable <Long> observable = Observable.interval（1，TimeUnit.SECONDS）

empty()

Observable.empty（）

empty（）运算符创建一个Observable，它不会发出任何项目但会正常终止，这在你需要为测试目的快速创建Observable时非常有用。

Observable <String> observable = Observable.empty();

附录

日常开发使用场景

1. 防止按钮重复点击；

<RxView.clicks(mClearContent).debounce(300, TimeUnit.MILLISECONDS)

2. EditText 添加 addTextChangedListener

RxTextView.textChanges(mUserNameEt)

3. 延迟执行

Observable.timer(3, TimeUnit.SECONDS)

4. 不断轮询

Observable.interval(3, 3, TimeUnit.SECONDS)

5. 减少频繁调用

RxTextView.textChanges(mUserNameEt)

6. RxPermission请求权限

rxPermissions.request(Manifest.permission.CAMERA)

RxJava深度剖析