RxJava 概述

a library for composing asynchronous and event-based programs using observable sequences for the Java VM”

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

概念

RxJava中观察者模式概念

Observable-可观察者,即被观察者

Observer-观察者

subscribe-订阅事件。Observable 和 Observer 通过 subscribe() 。

观察者/订阅者的回调事件

onNext(T t)正常的回调事件

onError(Throwable e) 异常发生时的回调事件

onCompleted() 完成时的回调事件

Observable 几种常见的构造方法

create(new Observable.OnSubscribe())

基础的构造方法

just(T…)

等同于onNext(t1,t2..)的调用

from(T[])

等同于

for(int i=0;i<i.length;i++){
onNext(t[i]);
}

的调用

示例

@SmallTest
public void testRxJava() {
//创建观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "completed");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
e.printStackTrace();
}
@Override
public void onNext(String s) {
Log.d(TAG, "get string ==" + s);
}
};
//创建订阅者 订阅者是观察者的一个实现类 封装了onStart() unsubscribe()等方法
Subscriber<String> stringSubscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "completed");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
e.printStackTrace();
}
@Override
public void onNext(String s) {
Log.d(TAG, "get string ==" + s);
}
};
//构造被观察者
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("simple");
subscriber.onNext("test");
subscriber.onNext("try");
}
});
//订阅观察者
Log.d(TAG, "observer subscribing");
observable.subscribe(observer);
observable.subscribe(observer);
}

Action

Action1、Action2、……、ActionN 是RxJava的一系列接口,只包含一个call()方法,Subscriber可以接受这一些列参数构造,产生不完整的回调。
示例

@MediumTest
public void testAction(){
Action0 action=new Action0() {
@Override
public void call() {
}
};
Action1<String> action1=new Action1<String>() {
@Override
public void call(String s) {
Log.d(TAG,"action1 get == "+ s);
}
};
Observable<String> observable=Observable.just("foo","bar","fooo","baar");
observable.subscribe(action1);
}

此外,还有三种方法重载

public final Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError)
public final Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onCompleted)

分别为支持onError、onComplted()的构造方式,可以看出,第二个方法和new Subscriber()已经并无区别了

Func

如同Action,Func1,Func2,…,FuncN为Rx提供的变换函数,所谓变换就是处理加工数据的过程,如把传入的URI转换成response或是将传入的资源转换为Bitmap

public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5, T t6)

为Observable设置Fuc的方法,一个具体的示例如下

@MediumTest
public void testFunc() {
Observable.just(45, 48, 78, 147, 88, 488)
.map(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
return String.valueOf(integer);
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.d(TAG, s);
}
});
}

此外,还有public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func)这个方法来设置将T 转换为R[]的一转多方法。

Scheduler

概述

Scheduler为RxJava的线程调度起,指定任务执行的线程

几个内置的Scheduler

  • Schedulers.immediate():不指定运行线程直接在当前线程运行,默认设置。
  • Schedulers.newThread(): 在新线程执行操作。
  • Schedulers .io(): I/O 线程,I/O操作所使用的 Scheduler。 的内部实现是是用一个无数量上限的线程池,可以重用空闲的线程,因此多数情况下 io() 比 newThread() 更有效率。
  • Schedulers.computation(): 计算所使用的Scheduler。这个计算指的是 CPU 密集型计算,即不会被 I/O 等操作限制性能的操作,使用大小为CUP核心数的的线程池。
  • AndroidSchedulers.mainThread():在Android 主线程即UI线程运行。

几个重要函数

public final Observable<T> observeOn(Scheduler scheduler)
public final Observable<T> subscribeOn(Scheduler scheduler)

分别为指定被观察者call()方法发生线程和订阅者事件回调线程。