欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页  >  php教程

RxJava操作符（三）Filtering

程序员文章站 2023-12-25 19:53:39
...

RxJava操作符（三）Filtering

在上一篇文章里，我们了解了转化操作符，能将数据转化为我们想要的格式，但是如果数据集合里面有一些我们想要过滤掉的数据怎么办？这时候我们就需要使用过滤操作符了，有点类似于sql里的where，让Observable只返回满足我们条件的数据。

一、debounce
debounce操作符就是起到了限流的作用，可以理解为阀门，当你半开阀门的时候，水会以较慢的速度流出来。不同之处就是阀门里的水不会浪费掉，而debounce过滤掉的数据会被丢弃掉。在Rxjava中，将这个操作符氛围了throttleWithTimeout和debounce两个操作符。先来看一下throttleWithTimeOut吧，如下图所示，这个操作符通过时间来限流，源Observable每次发射出来一个数据后就会进行计时，如果在设定好的时间结束前源Observable有新的数据发射出来，这个数据就会被丢弃，同时重新开始计时。如果每次都是在计时结束前发射数据，那么这个限流就会走向极端：只会发射最后一个数据。


首先我们来创建一个Observable，每隔100毫秒发射一个数据，当要发射的数据是3的倍数的时候，下一个数据就延迟到300毫秒再发射。

    

  1. private Observable createObserver() {
    

    2. 

  2. return Observable.create(new Observable.OnSubscribe() {
    

    3. 

  3. @Override
    

    4. 

  4. public void call(Subscriber super Integer> subscriber) {
    

    5. 

  5. for (int i = 0; i 
    6. 

  6. if (!subscriber.isUnsubscribed()) {
    

    7. 

  7. subscriber.onNext(i);
    

    8. 

  8. }
    

    9. 

  9. int sleep = 100;
    

    10. 

  10. if (i % 3 == 0) {
    

    11. 

  11. sleep = 300;
    

    12. 

  12. }
    

    13. 

  13. try {
    

    14. 

  14. Thread.sleep(sleep);
    

    15. 

  15. } catch (InterruptedException e) {
    

    16. 

  16. e.printStackTrace();
    

    17. 

  17. }
    

    18. 


    19. 

  19. }
    

    20. 

  20. subscriber.onCompleted();
    

    21. 

  21. }
    

    22. 

  22. }).subscribeOn(Schedulers.computation());
    

    23. 

  23. }
    24.
下面使用throttleWithTimeOut来过滤一下这个Observable，我们设定的过滤时间是200毫秒，也就说发射间隔小于200毫秒的数据会被过滤掉。

    

  1. private Observable throttleWithTimeoutObserver() {
    

    2. 

  2. return createObserver().throttleWithTimeout(200, TimeUnit.MILLISECONDS)
    

    3. 

  3. .observeOn(AndroidSchedulers.mainThread());
    

    4. 


    5. 

  5. }
    6.
对其进行订阅

    

  1. mLButton.setText("throttleWithTimeout");
    

    2. 

  2. mLButton.setOnClickListener(e -> throttleWithTimeoutObserver().subscribe(i -> log("throttleWithTimeout:" + i)));
    3.
运行结果如下，可以看到，不是3的倍数的数据在发射后200毫秒内会发射出新的数据，所以会被过滤掉。

debounce操作符也可以使用时间来进行过滤，这时它跟throttleWithTimeOut使用起来是一样，但是deounce操作符还可以根据一个函数来进行限流。这个函数的返回值是一个临时Observable，如果源Observable在发射一个新的数据的时候，上一个数据根据函数所生成的临时Observable还没有结束，那么上一个数据就会被过滤掉。

生成一个Observable并使用debounce对其进行过滤，只有发射来的数据为偶数的时候才会调用onCompleted方法来表示这个临时的Observable已经终止。

    

  1. private Observable debounceObserver() {
    

    2. 

  2. return Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).debounce(integer -> {
    

    3. 

  3. log(integer);
    

    4. 

  4. return Observable.create(new Observable.OnSubscribe() {
    

    5. 

  5. @Override
    

    6. 

  6. public void call(Subscriber super Integer> subscriber) {
    

    7. 

  7. if (integer % 2 == 0 && !subscriber.isUnsubscribed()) {
    

    8. 

  8. log("complete:" + integer);
    

    9. 

  9. subscriber.onNext(integer);
    

    10. 

  10. subscriber.onCompleted();
    

    11. 

  11. }
    

    12. 

  12. }
    

    13. 

  13. });
    

    14. 

  14. })
    

    15. 

  15. .observeOn(AndroidSchedulers.mainThread());
    

    16. 

  16. }
    17.
对其进行订阅

  1. mRButton.setOnClickListener(e -> debounceObserver().subscribe(i -> log("debounce:" + i)));
运行结果如下，可以看到，只有那些调用了onCompleted方法的数据才会被发射出来，其他的都过滤掉了。


二、Distinct
Distinct操作符的用处就是用来去重，非常好理解。如下图所示，所有重复的数据都会被过滤掉。还有一个操作符distinctUntilChanged,是用来过滤掉连续的重复数据。

创建两个Observable，并使用Distinct和DistinctUtilChanged操作符分别对其进行过滤

    

  1. private Observable distinctObserver() {
    

    2. 

  2. return Observable.just(1, 2, 3, 4, 5, 4, 3, 2, 1).distinct();
    

    3. 


    4. 

  4. }
    

    5. 


    6. 

  6. private Observable distinctUntilChangedObserver() {
    

    7. 

  7. return Observable.just(1, 2, 3, 3, 3, 1, 2, 3, 3).distinctUntilChanged();
    

    8. 


    9. 

  9. }
    10.
进行订阅

    

  1. mLButton.setText("distinct");
    

    2. 

  2. mLButton.setOnClickListener(e -> distinctObserver().subscribe(i -> log("distinct:" + i)));
    

    3. 

  3. mRButton.setText("UntilChanged");
    

    4. 

  4. mRButton.setOnClickListener(e -> distinctUntilChangedObserver().subscribe(i -> log("UntilChanged:" + i)));
    5.
运行结果如下所示，可以看到Distinct过滤掉了所有重复的数字，二DistinctUtilChanged只过滤掉重复的数字


三、ElementAt、Filter
这两个操作符都很好理解，ElementAt只会返回指定位置的数据，而Filter只会返回满足过滤条件的数据，其示意图分别如下所示

创建两个Observable对象并分别使用ElementAt和Filter操作符对其进行过滤

    

  1. private Observable elementAtObserver() {
    

    2. 

  2. return Observable.just(0, 1, 2, 3, 4, 5).elementAt(2);
    

    3. 

  3. }
    

    4. 


    5. 

  5. private Observable FilterObserver() {
    

    6. 

  6. return Observable.just(0, 1, 2, 3, 4, 5).filter(i -> i 
    7. 

  7. }
    8.
分别对其进行订阅

    

  1. mLButton.setText("elementAt");
    

    2. 

  2. mLButton.setOnClickListener(e -> elementAtObserver().subscribe(i -> log("elementAt:" + i)));
    

    3. 

  3. mRButton.setText("Filter");
    

    4. 

  4. mRButton.setOnClickListener(e -> FilterObserver().subscribe(i -> log("Filter:" + i)));
    5.
运行结果如下


四、First、Last
First操作符只会返回第一条数据，并且还可以返回满足条件的第一条数据。如果你看过我以前的博客，就会发现在我们使用Rxjava实现三级缓存的例子里，就是使用first操作符来选择所要使用的缓存。与First相反，Last操作符只返回最后一条满足条件的数据。

另外还有一个BlockingObservable方法,这个方法不会对Observable做任何处理，只会阻塞住，当满足条件的数据发射出来的时候才会返回一个BlockingObservable对象。可以使用Observable.toBlocking或者BlockingObservable.from方法来将一个Observable对象转化为BlockingObservable对象。BlockingObservable可以和first操作符进行配合使用。

创建两个Observable对象并分别使用first操作符进行处理

    

  1. private Observable FirstObserver() {
    

    2. 

  2. return Observable.just(0, 1, 2, 3, 4, 5).first(i -> i > 1);
    

    3. 

  3. }
    

    4. 


    5. 

  5. private BlockingObservable FilterObserver() {
    

    6. 

  6. return Observable.create(new Observable.OnSubscribe() {
    

    7. 

  7. @Override
    

    8. 

  8. public void call(Subscriber super Integer> subscriber) {
    

    9. 

  9. for (int i = 0; i 
    10. 

  10. try {
    

    11. 

  11. Thread.sleep(500);
    

    12. 

  12. } catch (InterruptedException e) {
    

    13. 

  13. e.printStackTrace();
    

    14. 

  14. }
    

    15. 

  15. if (!subscriber.isUnsubscribed()) {
    

    16. 

  16. log("onNext:" + i);
    

    17. 

  17. subscriber.onNext(i);
    

    18. 

  18. }
    

    19. 

  19. }
    

    20. 

  20. if (!subscriber.isUnsubscribed()) {
    

    21. 

  21. subscriber.onCompleted();
    

    22. 

  22. }
    

    23. 

  23. }
    

    24. 

  24. }).toBlocking();
    

    25. 

  25. }
    26.
分别进行订阅

    

  1. mLButton.setText("First");
    

    2. 

  2. mLButton.setOnClickListener(e -> FirstObserver().subscribe(i -> log("First:" + i)));
    

    3. 

  3. mRButton.setText(" Blocking");
    

    4. 

  4. mRButton.setOnClickListener(e -> {
    

    5. 

  5. log("blocking:" + FilterObserver().first(i -> i > 1));
    

    6. 

  6. });
    7.
运行结果如下。可以看到first操作符返回了第一个大于1的数2，而BlockingObservable则一直阻塞着，直到第一个大于1的数据发射出来。


五、Skip、Take
Skip操作符将源Observable发射的数据过滤掉前n项，而Take操作符则只取前n项，理解和使用起来都很容易，但是用处很大。另外还有SkipLast和TakeLast操作符，分别是从后面进行过滤操作。

创建两个Observable并分别使用skip和take操作符对其进行过滤操作

    

  1. private Observable skipObserver() {
    

    2. 

  2. return Observable.just(0, 1, 2, 3, 4, 5).skip(2);
    

    3. 

  3. }
    

    4. 

  4. private Observable takeObserver() {
    

    5. 

  5. return Observable.just(0, 1, 2, 3, 4, 5).take(2);
    

    6. 

  6. }
    7.
分别进行订阅

    

  1. mLButton.setText("Skip");
    

    2. 

  2. mLButton.setOnClickListener(e -> skipObserver().subscribe(i -> log("Skip:" + i)));
    

    3. 

  3. mRButton.setText("Take");
    

    4. 

  4. mRButton.setOnClickListener(e -> takeObserver().subscribe(i -> log("Take:" + i)));
    5.
运行结果如下，可以看到skip过滤掉了前两项，而take则过滤掉了除了前两项的其他所有项。


六、Sample、ThrottleFirst
Sample操作符会定时地发射源Observable最近发射的数据，其他的都会被过滤掉，等效于ThrottleLast操作符，而ThrottleFirst操作符则会定期发射这个时间段里源Observable发射的第一个数据

我们创建一个Observable每隔200毫秒发射一个数据，然后分别使用sample和throttleFirst操作符对其进行过滤

    

  1. private Observable sampleObserver() {
    

    2. 

  2. return createObserver().sample(1000, TimeUnit.MILLISECONDS);
    

    3. 

  3. }
    

    4. 


    5. 

  5. private Observable throttleFirstObserver() {
    

    6. 

  6. return createObserver().throttleFirst(1000, TimeUnit.MILLISECONDS);
    

    7. 

  7. }
    

    8. 


    9. 


    10. 

  10. private Observable createObserver() {
    

    11. 

  11. return Observable.create(new Observable.OnSubscribe() {
    

    12. 

  12. @Override
    

    13. 

  13. public void call(Subscriber super Integer> subscriber) {
    

    14. 

  14. for (int i = 0; i 
    15. 

  15. try {
    

    16. 

  16. Thread.sleep(200);
    

    17. 

  17. } catch (InterruptedException e) {
    

    18. 

  18. e.printStackTrace();
    

    19. 

  19. }
    

    20. 

  20. subscriber.onNext(i);
    

    21. 

  21. }
    

    22. 

  22. subscriber.onCompleted();
    

    23. 

  23. }
    

    24. 

  24. });
    

    25. 

  25. }
    26.
分别进行订阅

    

  1. mLButton.setText("sample");
    

    2. 

  2. mLButton.setOnClickListener(e -> sampleObserver().subscribe(i -> log("sample:" + i)));
    

    3. 

  3. mRButton.setText("throttleFirst");
    

    4. 

  4. mRButton.setOnClickListener(e -> throttleFirstObserver().subscribe(i -> log("throttleFirst:" + i)));
    5.
运行结果如下，可以看到sample操作符会每隔5个数字发射出一个数据来，而throttleFirst则会每隔5个数据发射第一个数据。


本文的demo程序见github:https://github.com/Chaoba/RxJavaDemo
相关标签： android

上一篇：

下一篇：

推荐阅读