本教程均是基于java的项目：
Create —使用 create() 方法来创建 Observable
private static void create() {
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
System.out.println("isDisposed: " + e.isDisposed());
if (!e.isDisposed()) {
e.onNext("aaa");
e.onNext("bbb");
e.onComplete();
e.onNext("bbb");
}
}
})
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe: " + d.isDisposed());
}
@Override
public void onNext(@NonNull String s) {
System.out.println("onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError: " + e.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
输出结果：
onSubscribe: false
isDisposed: false
onNext: aaa
onNext: bbb
onComplete
Just — 将一个对象或一组对象转换成Observable，并将他们发射出来（自动调用onNext()）
private static void just() {
Observable.just(1, 2, 3)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe: " + d.isDisposed());
}
@Override
public void onNext(@NonNull Integer s) {
System.out.println("onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError: " + e.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
输出结果：
onSubscribe: false
onNext: 1
onNext: 2
onNext: 3
onComplete
From — 将其他对象或数据结构转换成Observable。
注意：Rxjava2.0已经没有form，取而代之的是formArray，formCallable，formFuture等
private static void formArray() {
String[] array = new String[]{"123", "456", "789"};
Observable.fromArray(array)
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
System.out.println("onSubscribe: " + d.isDisposed());
}
@Override
public void onNext(@NonNull String s) {
System.out.println("onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
System.out.println("onError: " + e.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
输出结果：
onSubscribe: false
onNext: 123
onNext: 456
onNext: 789
onComplete
Defer — 直到有Observer订阅才创建Obervable，保证创建一个最新的Observable给Observer。
public static int i = 0;
private static void defer() {
i = 5;
Observable<Integer> just = Observable.just(i);
i = 10;
Observable<Integer> defer = Observable.defer(new Callable<ObservableSource<Integer>>() {
@Override
public ObservableSource<Integer> call() throws Exception {
return Observable.just(i);
}
});
i = 20;
just.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("just accept: " + integer);
}
});
defer.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("defer accept:" + integer);
}
});
i = 30;
just.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("just accept: " + integer);
}
});
defer.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println("accept:" + integer);
}
});
}
输出结果：
just accept: 5
defer accept:20
just accept: 5
accept:30
可以看到just输出一值是5，而defer会是最新的20，30。
Interval — 创建一个特定时间间隔的Observable
private static void interval() {
Observable
.interval(0, 1, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
System.out.println(Thread.currentThread() + " - " + aLong + " - " + System.currentTimeMillis());
}
});
while (true) ; //死循环 让程序一直运行
}
输出结果：
Thread[RxComputationThreadPool-1,5,main] - 0 - 1508660842333
Thread[RxComputationThreadPool-1,5,main] - 1 - 1508660843312
Thread[RxComputationThreadPool-1,5,main] - 2 - 1508660844312
Thread[RxComputationThreadPool-1,5,main] - 3 - 1508660845325
Thread[RxComputationThreadPool-1,5,main] - 4 - 1508660846325
注意 这里有个死循环，让程序一直运行才能看到效果，否则程序结束是看不到效果的。还有这个1s定时并没有相当准确（比如0-1之间只有979 差21ms），应用层也够了。
Timer —创建一个Observable延时后发送任务
private static void timer() {
System.out.println(Thread.currentThread() + " - " + System.currentTimeMillis());
final boolean[] flag = {true};
Observable
.timer(2, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
System.out.println(Thread.currentThread() + " - " + aLong + " - " + System.currentTimeMillis());
flag[0] = false;
}
});
while (flag[0]) ;
}
输出结果：
Thread[main,5,main] - 1508661300214
Thread[RxComputationThreadPool-1,5,main] - 0 - 1508661302392
Range — 创建一个Observable 发射一定范围的Integer
private static void range() {
Observable
.range(0,5)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println(integer);
}
});
}
输出结果：
0
1
2
3
4
Repeat — 创建一个Observable，重复发射一个特定的项目或者一系列项目
private static void repeat() {
Observable
.range(0,2)
.repeat(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
System.out.println(integer);
}
});
}
输出结果：
0
1
0
1
可以看到这个Range的Observable重复执行了2次
Start — 创建一个Observable 发射这个方法的返回值
注意：Rxjava2.0已经没有start，取而代之的是formCallable
private static void fromCallable() {
Observable
.fromCallable(new Callable<String>() {
@Override
public String call() throws Exception {
return "100";
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
}
输出结果：
100
Empty/Never/Throw — 创建Observables拥有精确的限制行为，Empty：创建一个不发射任何数据但是正常终止的Observable Never：创建一个不发射数据也不终止的Observable Throw：创建一个不发射数据以一个错误终止的Observable
网友评论