知识点补充_RxJava
推荐网址
http://gank.io/post/560e15be2dca930e00da1083
http://blog.csdn.net/qq_35064774/article/details/53057332
https://juejin.im/post/5848d96761ff4b0058c9d3dc
RxJava 到底是什么
响应式编程(Reactive programming)不是一种API,而是一种新的非常有用的范式,而RxJava就是一套基于此思想的框架,在Android开发中我们通过这个框架就能探索响应式的世界,同时结合另一个库,RxAndroid,这是一个扩展库,更好的兼容了Android特性,比如主线程,UI事件等。
简单理解:就是一个实现异步操作的库,而别的定语都是基于这之上的。
RxJava 好在哪
RxJava 的优势是使代码逻辑变得简洁
概念和介绍
扩展的观察者模式
观察者模式面向的需求是:A 对象(观察者)对 B 对象(被观察者)的某种变化高度敏感,需要在 B 变化的一瞬间做出反应。举例
新闻里喜闻乐见的警察抓小偷,警察需要在小偷伸手作案的时候实施抓捕。在这个例子里,- 警察是观察者
- 小偷是被观察者
观察为:警察需要时刻盯着小偷的一举一动,才能保证不会漏过任何瞬间。
- 程序的观察者模式和这种真正的『观察』略有不同,观察者不需要时刻盯着被观察者(例如 A 不需要每过 2ms 就检查一次B的状态),而是采用注册(Register)或者称为订阅(Subscribe)的方式,告诉被观察者:我需要你的某某状态,你要在它变化的时候通知我
Android 中的体现为:
监听器 OnClickListener
,对设置OnClickListener来说, View
是被观察者, OnClickListener
是观察者,二者通过setOnClickListener()
方法达成订阅关系。
当 View 接收到点击事件后,会将将点击事件发送给已经定阅了的 OnClickListener(回调)
- RxJava 的观察者模式
RxJava 有四个基本概念:Observable
(可观察者,即被观察者(小偷))、Observer
(观察者(警察))、subscribe
(订阅)、事件。
Observable
和Observer
通过subscribe()
方法实现订阅关系,从而Observable
可以在需要的时候发出事件来通知Observer
。
基本实现
创建 Observer(观察者)
它决定事件触发的时候将有怎样的行为。 RxJava 中的 Observer 接口的实现方式
Observer<String> observer = new Observer<String>() {
//被订阅
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.i(TAG, "onSubscribe: ");
}
@Override
public void onNext(@NonNull String s) {
Log.i(TAG, "onNext: ");
}
//在事件处理过程中出异常时会被触发
@Override
public void onError(@NonNull Throwable e) {
Log.i(TAG, "onError: ");
}
//onCompleted(): 事件队列完结。当不会再有新的 onNext() 发出时,需要触发 onCompleted() 方法作为标志。
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
}
};
创建 Observable(被观察者)
它决定什么时候触发事件以及触发怎样的事件
它的作用相当于一个计划表,当 Observable 被订阅的时候,OnSubscribe 的 call() 方法会自动被调用
这样,由被观察者调用了观察者的回调方法,就实现了由被观察者向观察者的事件传递,即观察者模式。
//被观察者
Observable observable = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(@NonNull ObservableEmitter emitter) throws Exception {
e.onNext("1111");
e.onNext("2222");
e.onNext("3333");
e.onComplete();
}
});
还可以使用便捷方法:just(T…): 将传入的参数依次发送出来。
ObservableEmitter
ObservableEmitter: Emitter是发射器的意思,这个就是用来发出事件的,它可以发出三种类型的事件,通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。
注意:
Observable
可以发出无限个onNext()
,Observer
也可以接受无限个onNext()
- 当有一个
onComplete()
发出后,Observer就不会接受onComplete()
之后的事件了 onError()
发出后,Observer
就不会接受到任何事件了- 可以不发送
onError()
或者onComplete()
Subscribe (订阅)
创建了 Observable 和 Observer 之后,再用 subscribe() 方法将它们联结起来,整条链子就可以工作了
observable.subscribe(observer);
// 或者:
//observable.subscribe(subscriber);
便捷方法使用
create() 方法是 RxJava 最基本的创造事件序列的方法。基于这个方法, RxJava还提供了一些方法用来快捷创建事件队列
- just(T…): 将传入的参数依次发送出来。
//just 方法会将传入的参数一次发送出来
// 将会依次调用:
// onNext("111");
// onNext("222");
// onNext("333");
// onCompleted();
Observable.just("111", "222", "333").subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull String s) {
Log.d(TAG, "onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});
- from(T[]) / from(Iterable) : 将传入的数组或 Iterable 拆分成具体对象后,依次发送出来。
// 传入的数组或 Iterable 拆分成具体对象后,依次发送出来。(有种遍历数组的感觉)
Observable.fromIterable(students).subscribe(new Observer<Student>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Student student) {
Log.d(TAG, "onNext: " + student.toString());
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});
- 不完整定义回调
subscribe()方法有多个重载
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) {}
最后一个带有Observer参数的我们已经使用过了
- 不带任何参数的 subscribe() 表示Observer(观察者)不关心任何事件,你Observable(被观察者)尽管发你的数据去吧,老子可不管你发什么。
- 带有一个Consumer参数的方法表示下游只关心onNext事件,其他的事件我假装没看见
因此我们如果只需要onNext事件可以这么写:
Consumer<Student> onNext = new Consumer<Student>() {
@Override
public void accept(@NonNull Student s) throws Exception {
Log.d(TAG, "accept: " + s.toString());
}
};
Observable<Student> studentObservable = Observable.fromIterable(students);
studentObservable.subscribe(onNext);
Consumer是简易版的Observer,他有多种重载,可以自定义你需要处理的信息,我这里调用的是只接受onNext消息的方法,
他只提供一个回调接口accept,由于没有主动设置onError和onCompete,接收到onCompete和onError之后做出默认操作。
使用RxAndroid 显示图片
普通使用
final Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Integer integer) {
iv.setImageResource(integer);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
};
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
observer.onNext(resId);
}
});
observable.subscribe(observer);
简化,使用匿名对象
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
e.onNext(resId);
e.onComplete();
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
iv.setImageResource(integer);
}
});
在 RxJava 的默认规则中,事件的发出和消费都是在同一个线程的。也就是说,如果只用上面的方法,实现出来的只是一个同步的观察者模式。观察者模式本身的目的就是『后台处理,前台回调』的异步机制,因此异步对于 RxJava 是至关重要的。而要实现异步,则需要用到 RxJava 的另一个概念: Scheduler 。
线程控制 - Scheduler
在不指定线程的情况下, RxJava遵循的是线程不变的原则
即:在哪个线程调用 subscribe()
,就在哪个线程生产事件;在哪个线程生产事件,就在哪个线程消费事件。如果需要切换线程,就需要用到 Scheduler (调度器)。
在RxJava 中,Scheduler
——调度器,相当于线程控制器,RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler ,它们已经适合大多数的使用场景:
使用方式和EvnetBus类似
- Schedulers.immediate(): 直接在当前线程运行,相当于不指定线程。这是默认的 Scheduler。
- Schedulers.newThread(): 总是启用新线程,并在新线程执行操作。
- Schedulers.io(): I/O 操作(读写文件、读写数据库、网络信息交互等)所使用的
- Scheduler。行为模式和 newThread() 差不多,区别在于 io()的内部实现是是用一个无数量上限的线程池,可以重用空闲的线程,因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中,可以避免创建不必要的线程。
- Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算,即不会被 I/O 等操作限制性能的操作,例如图形的计算。这个 Scheduler 使用的固定的线程池,大小为 CPU 核数。不要把 I/O 操作放在 computation() 中,否则 I/O 操作的等待时间会浪费 CPU。
另外, Android 还有一个专用的 AndroidSchedulers.mainThread(),它指定的操作将在 Android 主线程运行。
有了Scheduler,就可以通过 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了
- subscribeOn(): 指定 subscribe() 所发生的线程,即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
- observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
线程切换
Observer<String> consumer = new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull String s) {
Log.d(TAG, "accept: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
}
};
Observable<String> integerObservable = Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void
subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
e.onNext(requestNet());
}
});
integerObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(consumer);
//访问网络代码
public String requestNet() {
String requestUrl = "https://www.baidu.com";
// 新建一个URL对象
URL url = null;
try {
url = new URL(requestUrl);
HttpURLConnection urlConn = (HttpURLConnection) url.openConnection();
urlConn.setRequestMethod("GET");
urlConn.connect();
if (urlConn.getResponseCode() == 200) {
InputStream is = urlConn.getInputStream();
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] buffer = new byte[1024];
int len = 0;
while ((len = is.read(buffer)) != -1) {
baos.write(buffer, 0, len);
}
baos.close();
is.close();
byte[] byteArray = baos.toByteArray();
return new String(byteArray);
} catch (Exception e) {
Log.e(TAG, e.toString());
return null;
}
}
urlConn.disconnect();
} catch (MalformedURLException e) {
e.printStackTrace();
} catch (ProtocolException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
上面的代码网络数据将会发生在 IO 线程,而打印数据则被设定在了主线程。这就意味着,即使加载网络数据耗费了几十甚至几百毫秒的时间,也不会造成丝毫界面的卡顿。
变换(操作符)
RxJava 提供了对事件序列进行变换的支持,这是它的核心功能之一,所谓变换,就是将事件序列中的对象或整个序列进行加工处理,转换成不同的事件或事件序列。
需求一:使用RxJava打印学生名字
//学生数据
Student s1 = new Student();
s1.setName("李四");
ArrayList<String> c1 = new ArrayList<>();
c1.add("JAVA");
c1.add("ANDROID");
c1.add("HTML");
s1.setCourse(c1);
Student s2 = new Student();
s2.setName("王五");
ArrayList<String> c2 = new ArrayList<>();
c2.add("JAVA");
c2.add("HTML");
Student s3 = new Student();
s3.setName("赵六");
ArrayList<String> c3 = new ArrayList<>();
c3.add("JAVA");
c3.add("HTML");
students = new ArrayList<>();
students.add(s1);
students.add(s2);
students.add(s3);
//使用map取得学生的名字
/**
* 使用map取得学生的名字
* @param view
*/
public void bt9(View view) {
Observable.fromIterable(students).map(new Function<Student, String>() {
@Override
public String apply(@NonNull Student student) throws Exception {
return student.getName();
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept: "+ s);
}
});
}
更多操作符
filter 过滤,filter 是用于过滤数据的,返回false表示拦截此数据。
public void bt11(View view) {
Observable.fromIterable(students).filter(new Predicate<Student>() {
@Override
public boolean test(@NonNull Student student) throws Exception {
return student.getName().equals("李四");
}
}).subscribe(new Consumer<Student>() {
@Override
public void accept(@NonNull Student student) throws Exception {
Log.e(TAG, "accept: "+student.toString() );
}
});
}
如果只要前面两个数据,
Observable.fromIterable(students).take(2).subscribe(new Consumer<Student>() {
@Override
public void accept(@NonNull Student student) throws Exception {
Log.e(TAG, "accept: "+student.toString() );
}
});
如果我们想在订阅者接收到数据前干点事情,比如记录日志:
Flowable.just(1, 2, 3)
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("保存:" + integer);
}
})
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println(integer);
}
});
map()
变换
演示将图片资源转成bitmap对象
Observable.just(R.mipmap.ic_launcher).map(new Function<Integer, Bitmap>() {
@Override
public Bitmap apply(@NonNull Integer integer) throws Exception {
//加载图片
return BitmapFactory.decodeResource(getResources(), R.mipmap.ic_launcher);
}
}).subscribe(new Consumer<Bitmap>() {
@Override
public void accept(@NonNull Bitmap bitmap) throws Exception {
//显示图片
iv.setImageBitmap(bitmap);
}
});
转载自东哥笔记;