RxJava操作符之创建操作符(三)

前言

  通过前两篇文章对于RxJava概念,原理及使用的学习,想必各位码友已经基本掌握RxJava的逻辑与功能了,那么从这篇文章开始我们来研究RxJava的各类操作符。
  
  什么是操作符?通过之前的学习,我们发现Observable负责发送事件,Observer负责接收事件,而这个过程中想要对事件数据做出修改就需要交给操作符来负责啦。主流RxJava中操作符主要分为三类:转换操作符,过滤操作符,组合操作符。而我个人将用来创建Observable的操作符归为了一个新类型。本篇我们就来看看常用的创建操作符都有哪些以及如何使用。

创建操作符

Create

  上一篇文章的例子中我们已经实践了create操作符如何使用了,这里我们介绍一种便捷的创建Observer的方式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
for (int i = 0; i < 5; i++) {
subscriber.onNext("xulei" + i);
}
subscriber.onCompleted();
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("rx_test", "create:" + s);
}
});

  subscribe()的入参使用Action1代替原来的Observer,只需重写一个call()方法,等同于原Observer中onNext()方法。如果需要onComplete与onError状态,还可以如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("rx_test", "create:" + s);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
Log.e("rx_test", "onError:" + throwable.getMessage());
}
}, new Action0() {
@Override
public void call() {
Log.e("rx_test", "onCompleted");
}
});

  使用new ActionX代替new Observer,代码是不是看起来更加灵活与简洁呢。

Just

  just操作符可将某个或某些对象转化为Observable对象,并将其发射出去。参数可为一个或多个数字,字符串。也可为集合,数组,Iterate对象等。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Observable.just(1, 2, 3, 4, 5, 6).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rx_test", "just:数字:" + integer);
//数字或者字符串都是单个发射多次
}
});
List<String> stringList = new ArrayList<>();
stringList.add("Hello");
stringList.add("Ha");
stringList.add("RxJava");
Observable.just(stringList).subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> strings) {
Log.e("rx_test", "just:集合:" + strings.toString());
//集合或数组是直接发射集合整体,不会拆分
}
});

  输出结果:

1
2
3
4
5
6
7
just:数字:1
just:数字:2
just:数字:3
just:数字:4
just:数字:5
just:数字:6
just:集合:[Hello, Ha, RxJava]

From

  from操作符可将某个对象转化为Observable对象,并且将其发射出去。不同于just,他接收集合或数组,并可将集合数组遍历之后拆分发送。

1
2
3
4
5
6
7
8
9
10
List<String> stringList = new ArrayList<>();
stringList.add("Hello");
stringList.add("Ha");
stringList.add("RxJava");
Observable.from(stringList).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("rx_test", "from:" + s);
}
});

  输出结果:

1
2
3
from:Hello
from:Ha
from:RxJava

Range

  range(int start, int count)操作符,根据初始值start,与数量count,发射count次以start为基数依次增加的值。

1
2
3
4
5
6
Observable.range(4, 5).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rx_test", "range:" + integer);
}
});

  输出结果:

1
2
3
4
5
range:4
range:5
range:6
range:7
range:8

Defer

  defer操作符功能类似于just操作符,不同之处在于defer只有在调用subscribe()方法进行订阅时才创建Observable,而just操作符在初始化Observable就已经创建了,且只创建一个Observable实例。这里我们通过与just对比进行实践。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
Action1<String> action1 = new Action1<String>() {
@Override
public void call(String s) {
Log.e("rx_test", s);
}
};
//defer
Observable<String> deferObservable = Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
Object o = new Object();
return Observable.just("defer:hashCode:" + o.hashCode());
}
});
deferObservable.subscribe(action1);
deferObservable.subscribe(action1);
deferObservable.subscribe(action1);
//just
Observable<String> justObservable = Observable.just("just:hashCode:" + new Object().hashCode());
justObservable.subscribe(action1);
justObservable.subscribe(action1);
justObservable.subscribe(action1);

  输出结果:

1
2
3
4
5
6
defer:hashCode:112449879
defer:hashCode:118897732
defer:hashCode:191664429
just:hashCode:121878114
just:hashCode:121878114
just:hashCode:121878114

  由输出结果我们可以看出defer每次输出的Observable哈西值是不同的,说明其每subscribe订阅一次都会创建一个新的Observable,从而可保证Observable中的数据都是最新的。而just只有初始化的时候创建一次Observable。

Interval

  interval创建操作符,创建一个Observabel并每隔一段时间周期发射一个由0开始增加的数字。
  注意:此Observabel是运行在新的线程,所以更新UI需要在主线程中订阅

1
2
3
4
5
6
7
8
9
10
//每隔100ms发射一个数字,从0自增
Observable.interval(100, TimeUnit.MILLISECONDS) //单位为毫秒
.observeOn(AndroidSchedulers.mainThread())
.take(5) //取前5次事件发射,take为过滤操作符,后期会详细讲
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
Log.e("rx_test", "interval:" + aLong);
}
});

  输出结果:

1
2
3
4
5
6
7
8
9
10
100ms后...
interval:0
100ms后...
interval:1
100ms后...
interval:2
100ms后...
interval:3
100ms后...
interval:4

Timer

  timer操作符,创建一个Observable并隔一段时间后发射一个特殊的值,仅发射一次。
  注意:此Observabel是运行在新的线程,所以更新UI需要在主线程中订阅

1
2
3
4
5
6
7
8
9
//隔1s后发射一个数字
Observable.timer(1, TimeUnit.SECONDS) //单位为秒
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
Log.e("rx_test", "timer:" + aLong);
}
});

  输出结果:

1
2
1秒后...
timer:0

Delay

  delay操作符,可用于延迟一定时长再发送事件。

1
2
3
4
5
6
7
8
9
//延迟2秒后发射事件
Observable.just(1, 2, 3)
.delay(2, TimeUnit.SECONDS)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("rx_test", "delay:" + integer);
}
});

  输出结果:

1
2
3
4
2秒后...
delay:1
delay:2
delay:3

Repeat

  repeat(long count)操作符,将Observable重复发射count次。

1
2
3
4
5
6
7
8
//重复发射5次“Sherlock”
Observable.just("Sherlock").repeat(5)
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("rx_test", "repeat:" + s);
}
});

  输出结果:

1
2
3
4
5
repeat:Sherlock
repeat:Sherlock
repeat:Sherlock
repeat:Sherlock
repeat:Sherlock

  以上就是常用的一些创建操作符,类似但不常用的还有empty、never、error等等就不一一介绍了,有兴趣的码友可以自行百度。

线程调度Scheduler

  RxJava就是用来处理异步任务的,所以就牵扯到生产事件所在线程,处理事件所在线程的问题,下面来看一下RxJava提供的线程调度Scheduler都有哪些。


  

  有了Scheduler,RxJava当然也提供了方法来使用它们。
  .subscribeOn()指定被观察者Observable的执行线程。
  .observeOn()指定观察者Observer的执行线程。
  如第一篇文章中的例子:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
//获取要查询的小区集合
Observable.from(getCommunitiesFromServer())
.flatMap(new Func1<Community, Observable<House>>() {
@Override
public Observable<House> call(Community community) {
return Observable.from(community.getHouses());
}
})
.filter(new Func1<House, Boolean>() {
@Override
public Boolean call(House house) {
return house.getPrice() < 200;
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<House>() {
@Override
public void call(House house) {
//显示查询出来的房源信息
ShowSearchedHousesMessage();
}
});

  其中.subscribeOn(Schedulers.io())指定了Observable在io线程运行,通常用来执行从服务器获取数据,数据库加载等耗时操作。.observeOn(AndroidSchedulers.mainThread())指定了Observer在Android环境下的UI线程运行,通常用来获取到数据后进行UI刷新的操作。可根据实际需求选择不同线程类型。

总结

  到此,本篇关于RxJava的创建类操作符以及线程调度就讲解完毕了,下一篇我们将一起研究RxJava的四类操作符中的转换操作符都有哪些以及如何使用。
  技术渣一枚,有写的不对的地方欢迎大神们留言指正,有什么疑惑或者建议也可以在我Github上RxJavaDemo项目Issues中提出,我会及时回复。
  附上RxJavaDemo的地址:
  RxJavaDemo

坚持原创技术分享,您的支持是我前进的动力,谢谢!