两种创建Observable的方法（转）

转自：http://blog.csdn.net/nicolelili1/article/details/52038211

Observable.create()

create()方法使开发者有能力从头开始创建一个Observable。它需要一个OnSubscribe对象,这个对象继承Action1,当观察者订阅我们的Observable时，它作为一个参数传入并执行call()函数。

 

1 2 3 4 5 6 7

Observable.create(new Observable.OnSubscribe<Object>(){         @Override         public void call(Subscriber<? super Object> subscriber) {           } });

Observable通过使用subscriber变量并根据条件调用它的方法来和观察者通信。让我们看一个“现实世界”的例子：

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

Observable<Integer> observableString = Observable.create(new Observable.OnSubscribe<Integer>() {         @Override         public void call(Subscriber<? super Integer> observer) {             for (int i = 0; i < 5; i++) {                 observer.onNext(i);             }             observer.onCompleted();         } });   Subscription subscriptionPrint = observableString.subscribe(new Observer<Integer>() {     @Override     public void onCompleted() {         System.out.println("Observable completed");     }       @Override     public void onError(Throwable e) {         System.out.println("Oh,no! Something wrong happened！");     }       @Override     public void onNext(Integer item) {         System.out.println("Item is " + item);     } });

例子故意写的简单，是因为即便是你第一次见到RxJava的操作，我想让你明白接下来要发生什么。

我们创建一个新的Observable<Integer>,它执行了5个元素的for循环，一个接一个的发射他们，最后完成。

另一方面，我们订阅了Observable，返回一个Subscription
。一旦我们订阅了，我们就开始接受整数，并一个接一个的打印出它们。我们并不知道要接受多少整数。事实上，我们也无需知道是因为我们为每种场景都提供对应的处理操作：
* 如果我们接收到了整数，那么就打印它。
* 如果序列结束，我们就打印一个关闭的序列信息。
* 如果错误发生了，我们就打印一个错误信息。

Observable.from()

在上一个例子中，我们创建了一个整数序列并一个一个的发射它们。假如我们已经有一个列表呢？我们是不是可以不用for循环而也可以一个接一个的发射它们呢？

在下面的例子代码中，我们从一个已有的列表中创建一个Observable序列：

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

List<Integer> items = new ArrayList<Integer>(); items.add(1); items.add(10); items.add(100); items.add(200);   Observable<Integer> observableString = Observable.from(items); Subscription subscriptionPrint = observableString.subscribe(new Observer<Integer>() {     @Override     public void onCompleted() {         System.out.println("Observable completed");     }       @Override     public void onError(Throwable e) {         System.out.println("Oh,no! Something wrong happened！");     }       @Override     public void onNext(Integer item) {         System.out.println("Item is " + item);     } });

输出的结果和上面的例子绝对是一样的。

from()创建符可以从一个列表/数组来创建Observable,并一个接一个的从列表/数组中发射出来每一个对象，或者也可以从Java Future类来创建Observable，并发射Future对象的.get()方法返回的结果值。传入Future作为参数时，我们可以指定一个超时的值。Observable将等待来自Future的结果；如果在超时之前仍然没有结果返回，Observable将会触发onError()方法通知观察者有错误发生了。