observable 作为RxJava总要组成部分，observable 相当于观察者模式中的被观察者。下面详细介绍observable创建过程。
Create:返回一个observable 当执行Subscriber执行订阅，并执行特定的方法
<T> Observable<T> create(OnSubscribe<T> f)
Code:
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
if(!subscriber.isUnsubscribed){//判断是否取消订阅如果取消不会执行call()方法
subscriber.onNext("我很好");
subscriber.onCompleted();
}}});
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: " + s);
}};
执行：observable.subscribe(subscriber);
结果：
03-28 22:51:40.351 2062-2062/com.zline.rxjavatest I/MainActivity: onNext: 我很好
03-28 22:51:40.351 2062-2062/com.zline.rxjavatest I/MainActivity: onCompleted:
Defer:直到有订阅者订阅了才创建Observable 对象，才通过Observable工厂创建Observable并执行,这样确保Observable 每次状态为最新
<T> Observable<T> defer(Func0<Observable<T>> observableFactory)
code:
str="defer";
Observable<String> observable=Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
return Observable.just(str);
}});
str="defer 改变";
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: " + s);
}};
执行：
observable.subscribe(subscriber);
结果：
03-28 23:39:09.178 com.zline.rxjavatest I/MainActivity: onNext: defer 改变
03-28 23:39:09.178 com.zline.rxjavatest I/MainActivity: onCompleted:
注意:这里的Observable 时刻是最新的。运行值为“defer改变”而不是defer
Empty:返回Obervable 不执行任何其他操作，直接执行订阅者的onComplete()方法
<T>Observable<T> empty()
code:
Observable<String> observable=Observable.empty();
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: " + s);
}};
执行：
observable.subscribe(subscriber);
结果：
03-29 02:06:54.918 com.zline.rxjavatest I/MainActivity: onCompleted:
Never:产生一个不会执行任何参数永远不会结束的Observable，起初只用于测试。
Throw:创建一个不发射数据并且以错误结束的Observable。
From：将数组/列表用来创建Observable对象，将里面的对象一一当参数执行，或者可以用Future来创建Observable对象，将Future.get()的值作为参数执行，我们可以指定一个超时的值。Observable将等待来自Future的结果；如果在超时之前仍然没有结果返回，Observable将会触发onError()方法通知观察者有错误发生了。
<T> Observable<T> from(Future<? super T> future)
<T> Observable<T> from(Future<? super T> future,long timeout, TimeUnit unit)
<T> Observable<T> from(T[] array)
code:
public class FromCallback implements Callable<String> {
@Override
public String call() throws Exception {
String str="Future 测试";
return str;
}}
ExecutorService service= Executors.newSingleThreadExecutor();
Future future=service.submit(new FromCallback());
Observable<String> observable=Observable.from(future);//Future作为参数
String[] strs=newString[]{"Jack","Tom","Marry"};
Observable<String> observable=Observable.from(strs);//数组作为参数
Subscriber<String> subscriber=new Subscriber() {
@Override
public void onCompleted() {
Log.i(TAG,"onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.i(TAG,"onNext: "+ s);
}};
执行：
observable.subscribe(subscriber);
结果：
Future参数结果
03-29 03:05:21.023 /com.zline.rxjavatest I/MainActivity: onNext: Future 测试
03-29 03:05:21.023 /com.zline.rxjavatest I/MainActivity: onCompleted:
数组参数结果
03-29 03:24:57.250 /com.zline.rxjavatest I/MainActivity: onNext: Jack
03-29 03:24:57.250 /com.zline.rxjavatest I/MainActivity: onNext: Tom
03-29 03:24:57.250 /com.zline.rxjavatest I/MainActivity: onNext: Marry
03-29 03:24:57.250 /com.zline.rxjavatest I/MainActivity: onCompleted:
Interval：每隔一段时间就产生一个数字，这些数字从0开始，一次递增1直至无穷大
Observable<long> interval(long initialDelay,long period, TimeUnit unit)
code:
Observable<Long> observable=Observable.interval(1000,1000, TimeUnit.MILLISECONDS);
Subscriber<long> subscriber=new Subscriber() {
@Override
public void onCompleted() {
Log.i(TAG,"onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long aLong) {
Log.i(TAG,"onNext: "+ aLong);
}};
执行：
observable.subscribe(subscriber);
结果：
03-29 04:06:23.741 /com.zline.rxjavatest I/MainActivity: onNext: 0
03-29 04:06:24.735 /com.zline.rxjavatest I/MainActivity: onNext: 1
03-29 04:06:25.735 /com.zline.rxjavatest I/MainActivity: onNext: 2
03-29 04:06:26.736 /com.zline.rxjavatest I/MainActivity: onNext: 3
03-29 04:06:27.735 /com.zline.rxjavatest I/MainActivity: onNext: 4
03-29 04:06:28.734 /com.zline.rxjavatest I/MainActivity: onNext: 5
Just：把其他类型的对象和数据类型转化成Observable
<T>Observable<T> just(T t1,T t2,T t3)
code:
Observable<String> observable=Observable.just("Jack","Tom","Marry");
Action1<String> nextAction=new Action1<String>() {
@Override
public void call(String s) {
Log.i(TAG,"call: next Action::::"+ s);
}};
Action1<Throwable> erroAction=new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
Log.e(TAG,"call: erro Action"+ throwable.toString());
}};
Action0 completAction=new Action0() {
@Override
public void call() {
Log.i(TAG,"call: complate Action");
}};
执行：
observable.subscribe(nextAction,erroAction,complerAction);
结果：
03-29 04:31:10.268 com.zline.rxjavatest I/MainActivity: call: next Action:::Jack
03-29 04:31:10.268 com.zline.rxjavatest I/MainActivity: call: next Action:::Tom
03-29 04:31:10.268 com.zline.rxjavatest I/MainActivity: call: next Action:::Marry
03-29 04:31:10.268 com.zline.rxjavatest I/MainActivity: call: complate Action
Range:创建一组在从n开始，个数为m的连续数字的Observable,比如range(3,10)，就是创建3、4、5…12的一组数字。
Observable<Integer> range(int start,int count)
code:
Observable<Integer> observable=Observable.range(3,2);
Subscriber<Integer> subscriber=new Subscriber() {
@Override
public void onCompleted() {
Log.i(TAG,"onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.i(TAG,"onNext: "+ integer);
}};
执行：
observable.subscribe(subscriber);
结果：
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onNext: 3
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onNext: 4
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onCompleted:
Repeat:对某一个Observable，重复产生多次结果
Observable<T> repeat(final long count, Scheduler scheduler)
code:
Observable observable=Observable.range(3,2).repeat(2);
Subscriber subscriber=new Subscriber() {
@Override
public void onCompleted() {
Log.i(TAG,"onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.i(TAG,"onNext: "+ integer);
}};
执行：
observable.subscribe(subscriber);
结果：
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onNext: 3
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onNext: 4
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onNext: 3
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onNext: 4
03-29 04:41:23.266 com.zline.rxjavatest I/MainActivity: onCompleted:
Timer：创建一连串数字，间隔固定时间
Observable<Long> timer(long initialDelay,long period, TimeUnit unit, Scheduler scheduler)
code:
Observable<Long> observable=Observable.timer(1,TimeUnit.SECONDS);
Subscriber<Long> subscriber=newSubscriber() {
@Override
public void onCompleted() {
Log.i(TAG,"onCompleted: ");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long aLong) {
Log.i(TAG,"onNext: "+ aLong);
}};
执行：
observable.subscribe(subscriber);
结果：
03-29 06:00:20.155 com.zline.rxjavatest I/MainActivity: onNext: 0
03-29 06:00:20.155 com.zline.rxjavatest I/MainActivity: onCompleted
执行一次就结束
网友评论