引言
该篇文章主要是关于RxJava的创建操作符使用的代码讲解。关于创建被观察者操作符共有以下几种：
- create()
- just()
- fromArray()
- fromIterable()
- never()
- empty()
- error()
- defer()
- timer()
- interval()
- intervalRange()
- range()
- rangeLong()
1. create()
该操作符已经在文章中使用过了，关于其的具体使用，可以自己点击查看。RxJava实例-->基本使用
2.just()
快速创建被观察者对象（Observable），最多只能10个参数。
// 在创建后就会发送这些对象，相当于执行了onNext(1)、onNext(2)、onNext(3)、onNext(4)
Observable.just(1, 2, 3, 4,)
.subscribe(new Observer<Integer>() {
//观察者接收事件前，默认最先调用复写onSubscribe()
@Override
public void onSubscribe(Disposable d) {
Logger.e("开始采用subscribe连接");
}
//当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onNext(Integer value) {
Logger.e("对Next事件作出响应" + value);
}
//当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onError(Throwable e) {
Logger.e("对Error事件作出响应");
}
//当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onComplete() {
Logger.e("对Complete事件作出响应");
}
});
3. fromArray()
快速创建被观察者对象（Observable），能发送10个以上事件（数组形式）
//设置需要传入的数组
Integer[] items = {0,1,2,3,4,5,6};
//创建被观察者对象时传入数组
Observable.fromArray(items)
.subscribe(new Observer<Integer>() {
//观察者接收事件前，默认最先调用复写onSubscribe()
@Override
public void onSubscribe(Disposable d) {
Logger.e("数组遍历");
}
//当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onNext(Integer value) {
Logger.e("数组中的元素 = " + value);
}
//当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onError(Throwable e) {
Logger.e("对Error事件作出响应");
}
//当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onComplete() {
Logger.e("遍历结束");
}
});
4. fromIterable()
快速创建被观察者对象（Observable）发送10个以上事件（集合形式）
ArrayList<Integer> list = new ArrayList<>();
list.add(1);
list.add(2);
list.add(3);
list.add(4);
Observable.fromIterable(list)
.subscribe(new Observer<Integer>() {
//观察者接收事件前，默认最先调用复写onSubscribe()
@Override
public void onSubscribe(Disposable d) {
Logger.e("开始采用subscribe连接");
}
//当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onNext(Integer value) {
Logger.e("对Next事件作出响应" + value);
}
//当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onError(Throwable e) {
Logger.e("对Error事件作出响应");
}
//当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onComplete() {
Logger.e("对Complete事件作出响应");
}
});
5. never()
被观察者不发送任何事件。
6. empty()
被观察者仅发送complete事件。观察者会回调onComplete方法。
Observable.empty().subscribe(new Observer<Object>() {
//观察者接收事件前，默认最先调用复写onSubscribe()
@Override
public void onSubscribe(Disposable d) {
Logger.e("开始采用subscribe连接");
}
//当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onNext(Object value) {
Logger.e("对Next事件作出响应");
}
//当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onError(Throwable e) {
Logger.e("对Error事件作出响应");
}
//当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onComplete() {
Logger.e("对Complete事件作出响应");
}
});
log信息：
03-04 21:26:15.608 2844-2844/com.gjj.frame E/gjj: 开始采用subscribe连接
03-04 21:26:15.608 2844-2844/com.gjj.frame E/gjj: 对Complete事件作出响应
7. error()
被观察者仅发送error事件。
Observable.error(new RuntimeException()).subscribe(new Observer<Object>() {
//观察者接收事件前，默认最先调用复写onSubscribe()
@Override
public void onSubscribe(Disposable d) {
Logger.e("开始采用subscribe连接");
}
//当被观察者产生Next事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onNext(Object value) {
Logger.e("对Next事件作出响应");
}
//当被观察者产生Error事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onError(Throwable e) {
Logger.e("对Error事件作出响应");
}
//当被观察者产生Complete事件&观察者接受到时，会调用该复写方法进行相应
@Override
public void onComplete() {
Logger.e("对Complete事件作出响应");
}
});
log信息：
03-04 21:30:13.520 4530-4530/com.gjj.frame E/gjj: 开始采用subscribe连接
03-04 21:30:13.520 4530-4530/com.gjj.frame E/gjj: 对Error事件作出响应
8. defer()
直到有观察者（Observer）订阅时，才动态创建被观察对象（Observable）以及发送事件
//第一次对i的赋值
private int i= 10;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_defer);
//通过defer定义被观察者对象，此时观察者对象孩没创建
Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
@Override
public ObservableSource<? extends Integer> call() throws Exception {
return Observable.just(i);
}
});
//第二次对i的赋值
i =15;
//观察者开始订阅，此时才会调用defer()创建被观察者对象（Observable）
observable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG,"开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG,"接收到的整数是"+value);
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"对error事件做出响应");
}
@Override
public void onComplete() {
Log.d(TAG,"对Complete事件做出响应");
}
});
}
log信息：
06-14 11:03:27.138 7493-7493/com.gjj.frame D/DeferActivity: 开始采用subscribe连接
06-14 11:03:27.138 7493-7493/com.gjj.frame D/DeferActivity: 接收到的整数是15
06-14 11:03:27.138 7493-7493/com.gjj.frame D/DeferActivity: 对Complete事件做出响应
9. timer()
延迟指定时间后，发送一个数值0(Long类型)，一般用于检测
Observable.timer(4, TimeUnit.SECONDS)
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(Constant.TAG,"开始采用subscribe连接");
}
@Override
public void onNext(Long value) {
Log.d(Constant.TAG,"接收到的整数是"+value);
}
@Override
public void onError(Throwable e) {
Log.e(Constant.TAG,"对error事件做出响应");
}
@Override
public void onComplete() {
Log.d(Constant.TAG,"对Complete事件做出响应");
}
});
log信息：
06-14 11:11:19.802 8110-8110/com.gjj.frame D/RxJava: 开始采用subscribe连接
06-14 11:11:23.805 8110-8151/com.gjj.frame D/RxJava: 接收到的整数是0
06-14 11:11:23.806 8110-8151/com.gjj.frame D/RxJava: 对Complete事件做出响应
10. interval()
每隔指定时间就发送事件，发送事件的序列从0开始无限递增1的整数序列。
//参数说明：
//参数1：第一次延迟时间
//参数2：第二次延迟数字
//参数3：时间单位
Observable.interval(3,1, TimeUnit.SECONDS)
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(Constant.TAG,"开始采用subscribe连接");
}
@Override
public void onNext(Long value) {
Log.d(Constant.TAG,"接收到了事件"+value);
}
@Override
public void onError(Throwable e) {
Log.e(Constant.TAG,"对error事件做出响应");
}
@Override
public void onComplete() {
Log.d(Constant.TAG,"对Complete事件做出响应");
}
});
11. intervalRange()
每隔指定时间就发送事件，可指定发送的数据的数量，发送的事件序列从0开始、无限递增1的整数序列
//参数说明：
//参数1：事件序列起始点；
//参数2：事件数量
//参数3：第1次事件延迟发送时间
//参数4：时间间隔
//参数5：参数单位
Observable.intervalRange(3,10,2,1, TimeUnit.SECONDS)
//从3开始一共发送10个数据
//第一次延迟2s发送，之后每个1s产生一个数字（从0开始递增1）
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(Constant.TAG,"开始采用subscribe连接");
}
@Override
public void onNext(Long value) {
Log.d(Constant.TAG,"接收到了事件"+value);
}
@Override
public void onError(Throwable e) {
Log.e(Constant.TAG,"对error事件做出响应");
}
@Override
public void onComplete() {
Log.d(Constant.TAG,"对Complete事件做出响应");
}
});
log信息：
06-14 11:33:05.721 9141-9141/com.gjj.frame D/RxJava: 开始采用subscribe连接
06-14 11:33:07.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件3
06-14 11:33:08.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件4
06-14 11:33:09.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件5
06-14 11:33:10.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件6
06-14 11:33:11.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件7
06-14 11:33:12.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件8
06-14 11:33:13.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件9
06-14 11:33:14.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件10
06-14 11:33:15.723 9141-9238/com.gjj.frame D/RxJava: 接收到了事件11
06-14 11:33:16.724 9141-9238/com.gjj.frame D/RxJava: 接收到了事件12
06-14 11:33:16.727 9141-9238/com.gjj.frame D/RxJava: 对Complete事件做出响应
12. range()
连续发送一个事件，可指定范围
//参数说明：
//参数1：事件序列起始点
//参数2：事件数量
Observable.range(3,10)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(Constant.TAG,"开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(Constant.TAG,"接收到了事件"+value);
}
@Override
public void onError(Throwable e) {
Log.e(Constant.TAG,"对error事件做出响应");
}
@Override
public void onComplete() {
Log.d(Constant.TAG,"对Complete事件做出响应");
}
});
log信息：
06-14 12:12:55.184 10356-10356/com.gjj.frame D/RxJava: 开始采用subscribe连接
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件3
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件4
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件5
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件6
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 接收到了事件7
06-14 12:12:55.185 10356-10356/com.gjj.frame D/RxJava: 对Complete事件做出响应
13. rangeLong()
类似于rang,区别在于该方法支持数据类型long
参考文章：
