Create
——使用一个函数从头开始创建一个Observable
Create操作符
代码：
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
if (!subscriber.isUnsubscribed()) {//判断观察者的状态，如果没有订阅则不发送数据
subscriber.onNext("created 操作符1");
subscriber.onNext("created 操作符2");
subscriber.onNext("created 操作符3");
subscriber.onNext("created 操作符4");
subscriber.onCompleted();
}
}
}).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(String s) {
Log.d(TAG, s);
}
});
输出结果：
03-17 23:46:43.137 12009-12009/com.io.github.hjianfei.demo D/onResponse: created 操作符1
03-17 23:46:43.137 12009-12009/com.io.github.hjianfei.demo D/onResponse: created 操作符2
03-17 23:46:43.137 12009-12009/com.io.github.hjianfei.demo D/onResponse: created 操作符3
03-17 23:46:43.137 12009-12009/com.io.github.hjianfei.demo D/onResponse: created 操作符4
03-17 23:46:43.137 12009-12009/com.io.github.hjianfei.demo D/onResponse: onCompleted
From
——将其它种类的对象和数据类型转换为Observable
From操作符
代码：
String from[] = {"From操作符1", "From操作符2", "From操作符3", "From操作符4"};
Observable.from(from)
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(String s) {
Log.d(TAG, s);
}
});
结果：
03-18 00:02:39.591 28131-28131/com.io.github.hjianfei.demo D/onResponse: From操作符1
03-18 00:02:39.591 28131-28131/com.io.github.hjianfei.demo D/onResponse: From操作符2
03-18 00:02:39.591 28131-28131/com.io.github.hjianfei.demo D/onResponse: From操作符3
03-18 00:02:39.591 28131-28131/com.io.github.hjianfei.demo D/onResponse: From操作符4
03-18 00:02:39.591 28131-28131/com.io.github.hjianfei.demo D/onResponse: onCompleted
Interval
——创建一个按固定时间间隔发射整数序列的Observable
Interval
Interval操作符返回一个Observable，它按固定的时间间隔发射一个无限递增的整数序列。它接受一个表示时间间隔的参数和一个表示时间单位的参数。
interval(long,TimeUnit))
interval(long,TimeUnit,Scheduler))
代码：
Observable.interval(5, TimeUnit.SECONDS)
.subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(Long aLong) {
Log.d(TAG, aLong+"");
}
});
结果：
03-18 00:08:58.024 6346-6651/com.io.github.hjianfei.demo D/onResponse: 0
03-18 00:09:03.023 6346-6651/com.io.github.hjianfei.demo D/onResponse: 1
03-18 00:09:08.024 6346-6651/com.io.github.hjianfei.demo D/onResponse: 2
03-18 00:09:13.024 6346-6651/com.io.github.hjianfei.demo D/onResponse: 3
03-18 00:09:18.024 6346-6651/com.io.github.hjianfei.demo D/onResponse: 4
03-18 00:09:23.025 6346-6651/com.io.github.hjianfei.demo D/onResponse: 5
......
......
Just
——创建一个发射指定值的Observable
Just将单个数据转换为发射那个数据的Observable。
Just类似于From，但是From会将数组或Iterable的数据取出然后逐个发射，而Just只是简单的原样发射，将数组或Iterable当做单个数据。
Just操作符
代码：
Observable.just("Just操作符1", "Just操作符2", "Just操作符3", "Just操作符4")
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(String just) {
Log.d(TAG, just);
}
});
结果：
03-18 00:16:30.458 16301-16301/com.io.github.hjianfei.demo D/onResponse: Just操作符1
03-18 00:16:30.458 16301-16301/com.io.github.hjianfei.demo D/onResponse: Just操作符2
03-18 00:16:30.458 16301-16301/com.io.github.hjianfei.demo D/onResponse: Just操作符3
03-18 00:16:30.458 16301-16301/com.io.github.hjianfei.demo D/onResponse: Just操作符4
03-18 00:16:30.458 16301-16301/com.io.github.hjianfei.demo D/onResponse: onCompleted
Range
——创建一个发射特定整数序列的Observable
Range操作符发射一个范围内的有序整数序列，你可以指定范围的起始和长度。
RxJava将这个操作符实现为range函数，它接受两个参数，一个是范围的起始值，一个是范围的数据的数目。如果你将第二个参数设为0，将导致Observable不发射任何数据（如果设置为负数，会抛异常）。
Range操作符
代码：
Observable.range(1, 5)
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, integer + "");
}
});
结果：
03-18 00:19:55.669 22020-22020/com.io.github.hjianfei.demo D/onResponse: 1
03-18 00:19:55.669 22020-22020/com.io.github.hjianfei.demo D/onResponse: 2
03-18 00:19:55.669 22020-22020/com.io.github.hjianfei.demo D/onResponse: 3
03-18 00:19:55.669 22020-22020/com.io.github.hjianfei.demo D/onResponse: 4
03-18 00:19:55.669 22020-22020/com.io.github.hjianfei.demo D/onResponse: 5
03-18 00:19:55.669 22020-22020/com.io.github.hjianfei.demo D/onResponse: onCompleted
Repeat
——创建一个发射特定数据重复多次的Observable
Repeat重复地发射数据。某些实现允许你重复的发射某个数据序列，还有一些允许你限制重复的次数。
RxJava将这个操作符实现为repeat方法。它不是创建一个Observable，而是重复发射原始Observable的数据序列，这个序列或者是无限的，或者通过repeat(n)指定重复次数。
Repeat操作符
代码：
Observable.just("Repeat操作符1", "Repeat操作符2", "Repeat操作符3")
.repeat(2)
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(String just) {
Log.d(TAG, just);
}
});
结果：
03-18 00:24:50.660 28474-28474/com.io.github.hjianfei.demo D/onResponse: Repeat操作符1
03-18 00:24:50.661 28474-28474/com.io.github.hjianfei.demo D/onResponse: Repeat操作符2
03-18 00:24:50.661 28474-28474/com.io.github.hjianfei.demo D/onResponse: Repeat操作符3
03-18 00:24:50.665 28474-28474/com.io.github.hjianfei.demo D/onResponse: Repeat操作符1
03-18 00:24:50.665 28474-28474/com.io.github.hjianfei.demo D/onResponse: Repeat操作符2
03-18 00:24:50.665 28474-28474/com.io.github.hjianfei.demo D/onResponse: Repeat操作符3
03-18 00:24:50.665 28474-28474/com.io.github.hjianfei.demo D/onResponse: onCompleted
Timer
——创建一个Observable，它在一个给定的延迟后发射一个特殊的值。
Timer操作符创建一个在给定的时间段之后返回一个特殊值的Observable。
RxJava将这个操作符实现为timer函数。
timer返回一个Observable，它在延迟一段给定的时间后发射一个简单的数字0。
Timer操作符
代码：
Observable.timer(2, java.util.concurrent.TimeUnit.SECONDS)
.subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError:" + e.toString());
}
@Override
public void onNext(Long aLong) {
Log.d(TAG, aLong + "");
}
});
结果：
03-18 00:29:42.656 3065-3533/com.io.github.hjianfei.demo D/onResponse: 0
03-18 00:29:42.656 3065-3533/com.io.github.hjianfei.demo D/onResponse: onCompleted
还有更多的操作符请查看 RXJava官方文档
