一、RxJava的创建方式
image.png
二创建
1、create方式-普通创建
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
Log.i("RxJavaActivity", "该方法是否是在主线程中 subscribe：" + isMainThread());
e.onNext("is ok");
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("RxJavaActivity", "该方法是否是在主线程中 onSubscribe：" + isMainThread());
}
@Override
public void onNext(String value) {
Log.i("RxJavaActivity", "该方法是否是在主线程中 onNext：" + isMainThread());
}
@Override
public void onError(Throwable e) {
Log.i("RxJavaActivity", "该方法是否是在主线程中 onError：" + isMainThread());
}
@Override
public void onComplete() {
Log.i("RxJavaActivity", "该方法是否是在主线程中 onComplete：" + isMainThread());
}
});
2、just创建（特点：直接发送 传入的事件，最多可以发送10个）
// 1. 创建时传入整型1、2、3、4、5、6
// 在创建后就会发送这些对象，相当于执行了onNext(1)、onNext(2)、onNext(3)、onNext(4)、onNext(5)、onNext(6)
Observable.just(1, 2, 3,4,5,6)
// 至此，一个Observable对象创建完毕，以下步骤仅为展示一个完整demo，可以忽略
// 2. 通过通过订阅（subscribe）连接观察者和被观察者
// 3. 创建观察者 & 定义响应事件的行为
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
}
3、fromArray(发送10个以上事件---数组形式)
// 1. 设置需要传入的数组
Integer[] items = { 0, 1, 2, 3, 4 };
// 2. 创建被观察者对象（Observable）时传入数组
// 在创建后就会将该数组转换成Observable & 发送该对象中的所有数据
Observable.fromArray(items)
.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.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
}
4、fromIterable（直接发送 传入的集合List数据）
// 1. 设置一个集合
List<Integer> list = new ArrayList<>();
list.add(1);
list.add(2);
list.add(3);
// 2. 通过fromIterable()将集合中的对象 / 数据发送出去
Observable.fromIterable(list)
.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.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
5、defer（直到有观察者（Observer ）订阅时，才动态创建被观察者对象（Observable） & 发送事件）
<-- 1. 第1次对i赋值 ->>
Integer i = 10;
// 2. 通过defer 定义被观察者对象
// 注：此时被观察者对象还没创建
Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
@Override
public ObservableSource<? extends Integer> call() throws Exception {
return Observable.just(i);
}
});
<-- 2. 第2次对i赋值 ->>
i = 15;
<-- 3. 观察者开始订阅 ->>
// 注：此时，才会调用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.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
6、timer（延迟指定时间后，发送1个数值0----Long类型）
// 该例子 = 延迟2s后，发送一个long类型数值
Observable.timer(2, TimeUnit.SECONDS)
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Long value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
7、interval（每隔指定时间 就发送 事件）
// 参数说明：
// 参数1 = 第1次延迟时间；
// 参数2 = 间隔时间数字；
// 参数3 = 时间单位；
Observable.interval(3,1,TimeUnit.SECONDS)
// 该例子发送的事件序列特点：延迟3s后发送事件，每隔1秒产生1个数字（从0开始递增1，无限个）
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Long value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
8、intervalRange（每隔指定时间 就发送 事件，可指定发送的数据的数量）
// 参数说明：
// 参数1 = 事件序列起始点；
// 参数2 = 事件数量；
// 参数3 = 第1次事件延迟发送时间；
// 参数4 = 间隔时间数字；
// 参数5 = 时间单位
Observable.intervalRange(3,10,2, 1, TimeUnit.SECONDS)
// 该例子发送的事件序列特点：
// 1. 从3开始，一共发送10个事件；
// 2. 第1次延迟2s发送，之后每隔2秒产生1个数字（从0开始递增1，无限个）
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Long value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
9、range（连续发送 1个事件序列，可指定范围，类似于intervalRange（），但区别在于：无延迟发送事件）
// 参数说明：
// 参数1 = 事件序列起始点；
// 参数2 = 事件数量；
// 注：若设置为负数，则会抛出异常
Observable.range(3,10)
// 该例子发送的事件序列特点：从3开始发送，每次发送事件递增1，一共发送10个事件
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
10、rangeLong（类似于range（），区别在于该方法支持数据类型 = Long）
Observable.rangeLong(3,10)
// 该例子发送的事件序列特点：从3开始发送，每次发送事件递增1，一共发送10个事件
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Long value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
网友评论