RxJava----操作符：创建操作符

Creating Observables（创建操作符）

just

创建一个发射预定义好的数据的 Observable ，发射完这些数据后，事件流就结束了。

Observable.just(1,2,3)
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
log("Complete!");
}
@Override
public void onError(Throwable e) {
log(e.getMessage().toString());
}
@Override
public void onNext(Integer integer) {
log(integer+"");
}
});

结果：

1
2
3
Complete！

from

使用这个特殊的“创建”函数，我们可以从一个列表（可以了List或者Array）中创建一个Observable。Observable将发射出列表中的每一个元素，我们可以通过订阅它们来对这些发出的元素做出响应。

Integer[] items=new Integer[]{1,2,3};
List<Integer> list= Arrays.asList(items);
Observable.from(items)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
log(integer+"");
}
});

结果：

1
2
3

defer

defer 并没有定义一个新的 Observable， defer 只是用来声明当 Subscriber 订阅到一个 Observable 上时，该 Observable 应该如何创建。例如，如果我们想创建一个发射当前时间然后就结束的 Observable， 发射一个数据然后结束，看起来用 just 实现即可：

Observable<Long> now = Observable.just(System.currentTimeMillis());
now.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(String.valueOf(aLong));
}
});
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
now.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(String.valueOf(aLong));
}
});

结果：

1461721257817
1461721257817

注意上面两个 subscriber 相隔 1秒订阅这个 Observable，但是他们收到的时间数据是一样的！这是因为当订阅的时候，时间数据只调用一次。其实你希望的是，当 一个 subscriber 订阅的时候才去获取当前的时间。 defer 的参数是一个返回一个 Observable 对象的函数。该函数返回的 Observable 对象就是 defer 返回的 Observable 对象。重点是，每当一个新的 Subscriber 订阅的时候，这个函数就重新执行一次。

Observable<Long> now = Observable.defer(new Func0<Observable<Long>>() {
@Override
public Observable<Long> call() {
return Observable.just(System.currentTimeMillis());
}
});

now.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(String.valueOf(aLong));
}
});
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
now.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(String.valueOf(aLong));
}
});

结果：

1461721097102
1461721098103

range

range()方法用两个数字作为参数：第一个是起始点，第二个是我们想发射数字的个数。

Observable.range(5,3)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
log(integer+"");
}
});

结果：

5
6
7

interval

创建一个无限的计时序列，每隔一段时间发射一个数字，从 0 开始：



interval操作符默认情况下是运行在一个新线程上的，当然你可以通过传入参数来修改其运行的线程。

Subscription subscription = Observable.interval(3, TimeUnit.SECONDS)
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(aLong+"");
}
});
try {
System.in.read();
} catch (IOException e) {
e.printStackTrace();
}

结果：

0
1
2
...

如果我们不调用 unsubscribe 的话，这个序列是不会停止的。

上面的代码在最后有个 System.in.read(); 阻塞语句，这个语句是有必要的，不然的话，程序不会打印任何内容就退出了。原因是我们的操作不是阻塞的：我们创建了一个每隔一段时间就发射数据的 Observable，然后我们注册了一个 Subscriber 来打印收到的数据。这两个操作都是非阻塞的，而 发射数据的计时器是运行在另外一个线程的，但是这个线程不会阻止 JVM 结束当前的程序，所以 如果没有 System.in.read(); 这个阻塞操作，还没发射数据则程序就已经结束运行了。

timer

Observable.timer 有两个重载函数。timer操作符默认情况下是运行在一个新线程上的，当然你可以通过传入参数来修改其运行的线程。

第一个示例创建了一个 Observable， 该 Observable 等待一段时间，然后发射数据 0 ，然后就结束了。

Observable.timer(3,TimeUnit.SECONDS)
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(aLong+"");
}
});

结果：

0

它将3秒后发射0,然后就完成了。

另外一个示例是，先等待一段时间，然后开始按照间隔的时间一直发射数据:

Observable.timer(4,2,TimeUnit.SECONDS)
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
log(aLong+"");
}
});

结果：

0
1
2
...

上面的示例，先等待 4秒，然后每隔2秒开始发射数据。

empty

这个函数创建的 Observable 只发射一个 onCompleted 事件就结束了。

Observable<Integer> observable = Observable.empty();
Subscription subscription = observable.subscribe(new Subscriber<Integer>() {
public void onCompleted() {
log("Complete!");
}
@Override
public void onError(Throwable e) {
log(e.getMessage().toString());
}
@Override
public void onNext(Integer integer) {
log(integer+"");
}
});

结果：

Complete!

never

这个 Observable 将不会发射任何事件和数据。

Observable<Integer> observable = Observable.empty();
Subscription subscription = observable.subscribe(new Subscriber<Integer>() {
public void onCompleted() {
log("Complete!");
}
@Override
public void onError(Throwable e) {
log(e.getMessage().toString());
}
@Override
public void onNext(Integer integer) {
log(integer+"");
}
});

结果不打印任何数据

error

这个 Observable 将会发射一个 error 事件，然后结束。

Observable<Integer> observable =  Observable.error(new Exception("Oops"));
Subscription subscription = observable.subscribe(new Subscriber<Integer>() {
public void onCompleted() {
log("Complete!");
}
@Override
public void onError(Throwable e) {
log(e.getMessage().toString());
}
@Override
public void onNext(Integer integer) {
log(integer+"");
}
});

结果：

Oops

create

create 是非常强大的一个函数。可以创建任何你需要的 Observable。

static <T> Observable<T> create(Observable.OnSubscribe<T> f)

上面是 create 函数的定义，参数 Observable.OnSubscribe 看起来很简单。OnSubscribe 只有一个方法，该方法的参数为 Subscriber。在该方法内我们可以手工的发射事件和数据到 subscriber。

Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello RxJava");
subscriber.onCompleted();
}
});
Subscriber<String> subscriber = new Subscriber<String>() {
public void onCompleted() {
log("Complete!");
}
@Override
public void onError(Throwable e) {
log(e.getMessage().toString());
}
@Override
public void onNext(String s) {
log(s);
}
};
observable.subscribe(subscriber);

结果：

Hello RxJava
Complete！

在使用create操作符时，最好要在回调的call函数中增加isUnsubscribed的判断，以便在subscriber在取消订阅时不会再执行call函数中相关代码逻辑，从而避免导致一些意想不到的错误出现；

当有 Subscriber 订阅到这个 Observable 时（上面示例中的 observable ），这个 Subscriber 对象就是你实现的函数中的参数 Subscriber。然后你可以在你的代码中把数据发射到这个 subscriber 中。注意，当数据发射完后，你需要手工的调用 onCompleted 来表明发射完成了。

项目源码 GitHub求赞，谢谢！

引用：

RxJava 教程第二部分：事件流基础之 创建事件流 - 云在千峰

再多几个例子 | RxJava Essentials CN