RX定义
- Rx是一个函数库,让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序
- Rx是微软.NET的一个响应式扩展。Rx借助可观测的序列提供一种简单的方式来创建异步的,基于事件驱动的程序。
- Rx就是一种响应式编程,来创建基于事件的异步程序
RXJava简介
RxJava就是一种用Java语言实现的响应式编程,来创建基于事件的异步程序
官方定义:一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库
RXJava原理
- RxJava: 基于 一种扩展的观察者模式
- RxJava的扩展观察者模式中有4个角色:
简单使用方法
代码示例:
Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onComplete(); } }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer() { @Override public void accept(Integer integer) throws Exception { Log.d(TAG, “accept integer = ” + integer); } });
解释说明:
- 创建被观察者,发送事件,即emitter.onNext(1),发送了数字1;
- 指定被观察者的执行线程,Schedulers.io();
- 观察者处理接收到的数据;
- 切换线程到主线程,观察者将在主线程处理这些数据。
实战项目解析
添加依赖
compile ‘io.reactivex.rxjava2:rxjava:2.0.0-RC5’compile ‘io.reactivex.rxjava2:rxandroid:2.0.0-RC1’
基理
Observable和Observer通过subscribe()方法实现订阅关系; Rxjava中是**自动发送事件**的, 一旦**订阅**就开始**发送**;
基本使用三个步骤
- 第一步,创建Observable, Emitter是发射器的意思, 在subscribe方法中不断调用发射器的方法; 总共有onNext()、onComplete()、onError()三个方法; 用法参考Rxjava1:
- 第二步,创建Observer, 总共有onNext()、onComplete()、onError()、onSubscribe()四个方法; 其中,**onNext()、onComplete()、onError()**三个方法分别对应着第一步中**Observable**的**onNext()、onComplete()、onError()**三个方法, 只要**Observable**发出(调用)对应的方法, **Observer**对应的方法就会被调用;
- onError()和onComplete是互斥的,一次只能调用一个;
- 第三步,订阅, observable.subscribe(observer);
下面开始实战内容
- 在app/build.gradle中添加依赖:
dependencies { implementation fileTree(dir: ‘libs’, include: [‘*.jar’]) implementation ‘com.android.support:appcompat-v7:28.0.0’ implementation ‘io.reactivex.rxjava2:rxjava:2.0.0-RC5’ implementation ‘io.reactivex.rxjava2:rxandroid:2.0.0-RC1’ implementation ‘com.squareup.retrofit2:retrofit:2.1.0’ implementation ‘com.squareup.retrofit2:converter-gson:2.1.0’// compile ‘com.squareup.retrofit2:adapter-rxjava:2.1.0’// compile ‘com.jakewharton.rxbinding:rxbinding:0.4.0’}
activity_main.xml:
MainActivity,java: 第一步,通过create()创建Observable(模拟对象:程序员), 通过onNext()发送数据:
public Observable getObservable() { return Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter e) throws Exception { e.onNext(“奏笛”); e.onNext(“泡吧”); e.onComplete(); } }); }
第二步,创建Observer(模拟对象:程序员女朋友), 创建的方法是直接new:
public Observer getObserver(){ return new Observer() { @Override public void onSubscribe(Disposable d) { Log.d(“MainActivity”,”onSubscribe”); } @Override public void onNext(String value) { Log.d(“MainActivity”,”onNext”); } @Override public void onError(Throwable e) { Log.d(“MainActivity”,”onError”); } @Override public void onComplete() { Log.d(“MainActivity”,”onComplete”); } }; }
第三步,在click()中, getObservable和observer, 实现订阅:
public void click(View view) { Observable observale = getObservable(); Observer observer = getObserver(); observale.subscribe(observer); }
以上便完成了一个最基本的使用; 运行效果:
点击按钮后打印日志:
由此可以应证, Rxjava中是自动发送事件的, 一旦Observable 被 observer 订阅了(observale.subscribe(observer);), Observable就开始发送; 由Observable通过自身ObservableOnSubscribe中的subscribe()中的 onNext()等方法自动发出信息, observer接收到信息后执行对应的onNext()等方法; 在订阅之后,Observer中, onSubscribe()是每次接收数据之前必须要调用的方法; onNext()则是对应Observable调用的次数去调用相应的次数; onComplete()、onError()对应完成/异常状态时候调用;
@Override public void onSubscribe(Disposable d) { Log.d(“MainActivity”,”onSubscribe”); }
接下来关注一下Observer构造方法中的onSubscribe()方法;
其中注意参数Disposable d,
Disposable是一次性的意思; 其主要有以下两个方法:
用法示例 (用于监听Observable发送的数据, 如果Observable发送的数据等于某个值, 就断绝订阅关系):
更改Observable代码:
public Observable getObservable() { return Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter e) throws Exception { e.onNext(“奏笛”); e.onNext(“泡吧”); e.onNext(“酗酒”); e.onComplete(); } }); }
运行示例,点击按钮:
可以发现已经没有onComplete()方法的打印信息了, 因为在onNext()中途已经断绝订阅关系了;
另外还有省略observer的简洁写法
public void click(View view) { Observable observale = getObservable(); observale.subscribe(new Consumer() { @Override public void accept(String s) throws Exception { if(s.equals(“奏笛”)){ Log.d(“MainActivity”,”收到奏笛!!这里类似于Observer的onNext()”); } if(s.equals(“吟诗”)){ Log.d(“MainActivity”,”收到吟诗!!这里类似于Observer的onNext()”); } if(s.equals(“酗酒”)){ Log.d(“MainActivity”,”收到酗酒!!这里类似于Observer的onNext()”); } } }, new Consumer() { @Override public void accept(Throwable throwable) throws Exception { Log.d(“MainActivity”,”这里类似于Observer的onError()”); } }, new Action() { @Override public void run() throws Exception { Log.d(“MainActivity”,”这里类似于Observer的onComplete()”); } }); } public Observable getObservable() { return Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter e) throws Exception { e.onNext(“奏笛”); e.onNext(“吟诗”); e.onNext(“酗酒”); e.onComplete(); } }); }
运行示例,点击按钮:
更改上面代码,
用Observable.just()创建Observable对象,
效果也是一样的; 运行示例,点击按钮,打印日志同上:
public Observable getObservable() { Observable observable = Observable.just(“奏笛”,”吟诗”,”酗酒”); return observable; }
或者显示在TextView上:
用Observable.fromArray()创建Observable对象,
Observable observable = Observable.fromArray(“奏笛”,”泡吧”,”吟诗”); 其实用法跟**just()**是一样的; **just()**源码如下, 里面最终也是调用**fromArray()**实现的:
用Observable.fromCallable()创建Observable对象
特点:只能返回一个数据;
本节笔记Activity全文(注意io.reactivex包的引用):
package com.cniao5.cniao5rxjava2demo;import android.support.v7.app.AppCompatActivity;import android.os.Bundle;import android.util.Log;import android.view.View;import android.widget.TextView;import java.util.concurrent.Callable;import io.reactivex.Observable;import io.reactivex.ObservableEmitter;import io.reactivex.ObservableOnSubscribe;import io.reactivex.Observer;import io.reactivex.disposables.Disposable;import io.reactivex.functions.Action;import io.reactivex.functions.Consumer;public class MainActivity extends AppCompatActivity { private TextView textView; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); textView = (TextView) findViewById(R.id.text); } public void click(View view) { Observable observale = getObservable();// Observer observer = getObserver();//// observale.subscribe(observer);// observale.subscribe(new Consumer() {// @Override// public void accept(String s) throws Exception {// Log.d(“MainActivity”,”accept=”+s);//// textView.append(s);// textView.append(“//n”);// }// }); observale.subscribe(new Consumer() { @Override public void accept(String s) throws Exception { if(s.equals(“奏笛”)){ Log.d(“MainActivity”,”收到奏笛!!这里类似于Observer的onNext()”); } if(s.equals(“吟诗”)){ Log.d(“MainActivity”,”收到吟诗!!这里类似于Observer的onNext()”); } if(s.equals(“酗酒”)){ Log.d(“MainActivity”,”收到酗酒!!这里类似于Observer的onNext()”); } textView.append(s); textView.append(“”); } }, new Consumer() { @Override public void accept(Throwable throwable) throws Exception { Log.d(“MainActivity”,”这里类似于Observer的onError()”); } }, new Action() { @Override public void run() throws Exception { Log.d(“MainActivity”,”这里类似于Observer的onComplete()”); } }); } public Observable getObservable() {// Observable observable = Observable.just(“奏笛”,”吟诗”,”酗酒”);// Observable observable = Observable.fromArray(“奏笛”,”泡吧”,”吟诗”); return Observable.fromCallable(new Callable() { @Override public String call() throws Exception { return “奏笛”; } });// return observable;// return Observable.create(new ObservableOnSubscribe() {// @Override// public void subscribe(ObservableEmitter e) throws Exception {// e.onNext(“奏笛”);// e.onNext(“吟诗”);// e.onNext(“酗酒”);// e.onComplete();//// e.onError(new);//// }// }); } public Observer getObserver(){ return new Observer() { Disposable dd =null;//定义一个变量局部变量 @Override public void onSubscribe(Disposable d) { dd = d;//把这段订阅关系的Disposable变量拿下来 Log.d(“MainActivity”,”onSubscribe”); } @Override public void onNext(String value) { Log.d(“MainActivity”,”onNext”); if(value.equals(“酗酒”)){ dd.dispose();//如果发送的数据等于某个值,就断绝关系 Log.d(“MainActivity”,”你的小可爱已经不想理你了!!!”); } } @Override public void onError(Throwable e) { Log.d(“MainActivity”,”onError”); } @Override public void onComplete() { Log.d(“MainActivity”,”onComplete”); } }; }}
小结
以上是我在项目中使用Rxjava的一些介绍和使用心得。以及由此引出的引入三方库之后,我们应该如何使用和学习的一些粗浅看法。Android架构师的学习进阶路线及资料,需要可以私信我发送“核心笔记”或“手册”即可获取!
一句话,就是在引入一个库后,能够学习它背后的原理。