Observable详解一(Observable创建)

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

执行一次就结束