RxJava(二)-这么庞大的家族?

RxJava(二)-这么庞大的家族?

在前面一章节中,讲解了Rxjava的基本的组成,他们之间的关系,这一章,主要让大家了解一下在RxJava中的操作符。但是在这之前,我们还得先了解一下 Scheduler(调度器)

Scheduler(调度器)

在上一章节,我们讲到了,Rxjava的就是异步
那么如何让他进行异步呢?这就要用到我们的调度器了。
先看看RxJava中调度器支持哪几种调度:

调度器类型 效果
Schedulers.computation( ) 用于计算任务,如事件循环或和回调处理,不要用于IO操作(IO操作请使用Schedulers.io());默认线程数等于处理器的数量
Schedulers.from(executor) 使用指定的Executor作为调度器
Schedulers.immediate( ) 在当前线程立即开始执行任务
Schedulers.io( ) 用于IO密集型任务,如异步阻塞IO操作,这个调度器的线程池会根据需要增长;对于普通的计算任务,请使用Schedulers.computation();Schedulers.io( )默认是一个CachedThreadScheduler,很像一个有线程缓存的新线程调度器
Schedulers.newThread( ) 为每个任务创建一个新线程
Schedulers.trampoline( ) 当其它排队的任务完成后,在当前线程排队开始执行

有了上述所说的几种调度之后,就可以调用subscribeOn()和observeOn()来对线程进行调度了。
subscribeOn()指定:Observable将全部的处理过程(包括发射数据和通知)放在特定的调度器上执行。
ObserveOn()指定:一个Observable在一个特定的调度器上调用观察者的onNext, onError和onCompleted方法,

Subscriber subcriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted: Completed!");
}

@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: Error!");
}

@Override
public void onNext(String s) {
Log.d(TAG, "onNext: " + s);
}
};

Observable.just("1", "2", "3", "4")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subcriber);

上面这段代码中,由于指定了1,2,3,4发射代码为Schedulers.io(),那么发射数据就将在io线程中执行。而onNext, onError和onCompleted则将在主线中执行。

Operators(操作符)

map家族

RxJava提供了几个mapping函数:map(),flatMap(),concatMap(),flatMapIterable()以及switchMap().所有这些函数都作用于一个可观测序列,然后变换它发射的值,最后用一种新的形式返回它们。

map

map 是用于变换的一个操作符,这在RxJava中占据了一定的地位,就是因为它的变换操作。

Subscriber subcriber = new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted: Completed!");
}

@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: Error!");
}

@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
};

Observable.just("1", "2", "3", "4")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return Integer.parseInt(s);
}
})
.subscribe(subcriber);

在上面的代码中,我通过map将字符串转化成了整形的1,2,3,4,返回一个Observable的对象。
请注意:这个操作符默认不在任何特定的调度器上执行。

flatmap

flatmap对于新入门的来说,理解起来确实有一定的难度,可以先看一个简单的栗子:

Subscriber subcriber = new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted: Completed!");
}

@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: Error!");
}

@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
};

Observable.just("1", "2", "3", "4")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<String, Observable<Integer>>() {
@Override
public Observable<Integer> call(String s) {
return Observable.just(Integer.parseInt(s)+1);
}
})
.subscribe(subcriber);

从上面我们可以看出,map与flatmap很相似,都是用的Func1,而且模式都是<I,O>模式,即是:I转换成O并返回。但是最大的不同点在于:我们flatmap的输出类型是Observable的类型。
在这里请注意一个问题:在执行flatmap中返回之后(O输出返回的Observable),并不是立马把返回的Observable通过Subscribe进行订阅,而是将返回的若干Observables都交给同一个Observable,然后再进行subscribe。

所以,在上面我们先将字符串”1”,”2”, “3”, “4” 分别转换成一个整形的Observable类型,即是:Observable(2),Observable(3),Observable(4),Observable(5)。然后将这些个Observables统一转换成一个Observable,再进行subscribe。看一下结果:

onNext: 2
onNext: 3
onNext: 4
onNext: 5
onCompleted: Completed!

那么,这个flatmap到底有何用呢?可以用在什么地方呢?

假设这样一种情景:一个学校的老师我们定义为一个集合A,每个老师包括了个人信息和所教课程,一个老师不可能只教授一门课程,所以我们将老师所教授课程定义为集合B。如果让你打印每个老师所教课程,该怎么做?

Teacher[] teachers = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() {
@Override
public void onNext(Course course) {
Log.d(tag, course.getName());
}
...
};
Observable.from(teachers)
.flatMap(new Func1<Teacher, Observable<Course>>() {
@Override
public Observable<Course> call(Teacher teacher) {
return Observable.from(teacher.getCourses());
}
})
.subscribe(subscriber);

最后再补充一点:FlatMap对这些Observables发射的数据做的是合并(merge)操作,因此它们可能是交错的。这意味着flatMap()函数在最后的Observable中不能够保证源Observables确切的发射顺序。

ConcatMap

RxJava的concatMap()函数解决了flatMap()的交叉问题,提供了一种能够把发射的值连续在一起的铺平函数,而不是合并它们,如下图所示:
ConcatMap示意图
变换的操作符还有很多:buffer,Scan…等等,大家可以研究一下。

其他操作符

repeat

让你发射的数据重复发射

Subscriber subcriber = new Subscriber<Integer>() {
...
}
};

Observable.just("1", "2","3")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<String, Observable<Integer>>() {
@Override
public Observable<Integer> call(String s) {
return Observable.just(Integer.parseInt(s)+1);
}
})
.repeat(3)
.subscribe(subcriber);

看一下结果:

onNext: 2
onNext: 3
onNext: 4
onNext: 2
onNext: 3
onNext: 4
onNext: 2
onNext: 3
onNext: 4
onCompleted: Completed!
range

从起始点开始发射数据

Subscriber subcriber = new Subscriber<Integer>() {
...
};

Observable.range(10,3)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subcriber);

结果为:10,11,12。range(10,3),其中10 是起始,3是数量。

interval

在需要轮询的时候是最好的选择

Observable.interval(3,TimeUnit.SECONDS)
        .subscribe(new Observer<Long>() {
            @Override
            public void onCompleted() {
            ...
        });

interval()函数的两个参数:一个指定两次发射的时间间隔,另一个是用到的时间单位。

take
Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
          .take(4)
          .subscribe(new Subscriber<Integer>() {
        ...
    });

输出

Next: 1
Next: 2
Next: 3
Next: 4
Sequence complete.
TakeLast

如果我们想要最后N个元素,我们只需使用takeLast()函数:

Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
              .takelast(2)
              .subscribe(new Subscriber<Integer>() {
            ...
        });

输出

Next: 7
Next: 8
Sequence complete.

当然Rxjava的操作符不止这一点,大家可以戳我可以查看更多的操作符的用法。

Rx系列文章:
第一篇《Rx系列之RxJava初识》
第三篇《Rx系列之Rxjava操作符进阶-使用场景》


 

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×