RXJava学习笔记（2）

2023年11月5日 90次阅读来源: 皮球二二

变换操作

通过上篇的学习，我们知道了如何去创建一个Observable对象，通过subscribe将事件传递给Observer，但是仅仅知道这些是不够的，有些复杂的场景，需要我们对Observable进行一些转换来发射数据。那么我们就在这篇文章里来系统的学习一下吧

Buffer操作符

Observable.from(email).buffer(3).subscribe(
new Action1<List<String>>() {
      @Overridepublic void call(List<String> strings) { 
            for (int i=0;i<strings.size();i++) {
                 Log.d("SampleTransformingActiv", "strings:" + strings.get(i)); 
            } 
            Log.d("SampleTransformingActiv", "==========================");
      }
});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: strings:email1
com.renyu.rxdemo D/SampleTransformingActiv: strings:email2
com.renyu.rxdemo D/SampleTransformingActiv: strings:email3
com.renyu.rxdemo D/SampleTransformingActiv: ==========================
com.renyu.rxdemo D/SampleTransformingActiv: strings:email4
com.renyu.rxdemo D/SampleTransformingActiv: strings:email5
com.renyu.rxdemo D/SampleTransformingActiv: strings:email6
com.renyu.rxdemo D/SampleTransformingActiv: ==========================
com.renyu.rxdemo D/SampleTransformingActiv: strings:email7
com.renyu.rxdemo D/SampleTransformingActiv: strings:email8
com.renyu.rxdemo D/SampleTransformingActiv: strings:email9
com.renyu.rxdemo D/SampleTransformingActiv: ==========================
com.renyu.rxdemo D/SampleTransformingActiv: strings:email10
com.renyu.rxdemo D/SampleTransformingActiv: ==========================

buffer就是将数据按照规定的数量做一个缓存，达到这个数量之后再一次性全部发射出去，在本例中，就是每缓存3个之后再发射出去再看下2个参数的方法

Observable.from(email).buffer(3, 5).subscribe(
      new Action1<List<String>>() { 
            @Override public void call(List<String> strings) { 
                for (int i=0;i<strings.size();i++) { 
                    Log.d("SampleTransformingActiv", "strings:" + strings.get(i)); 
                } 
                Log.d("SampleTransformingActiv", "==========="); 
            }
 });

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: strings:email1
com.renyu.rxdemo D/SampleTransformingActiv: strings:email2
com.renyu.rxdemo D/SampleTransformingActiv: strings:email3
com.renyu.rxdemo D/SampleTransformingActiv: ===========
com.renyu.rxdemo D/SampleTransformingActiv: strings:email6
com.renyu.rxdemo D/SampleTransformingActiv: strings:email7
com.renyu.rxdemo D/SampleTransformingActiv: strings:email8
com.renyu.rxdemo D/SampleTransformingActiv: ===========

buffer(count, skip)，代表每发射skip个数据用count数据作为缓存，再进行发射，如果count==skip，就是第一种情况了`buffer不仅在数量上可以缓存，时间上也同样可以控制缓存

subscription=Observable.create(
    new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { 
        if (!subscriber.isUnsubscribed()) { 
            for (int i=0;i<email.length;i++) { 
                subscriber.onNext(email[i]); 
                try { 
                      Thread.sleep(500); 
                } catch (InterruptedException e) { 
                      e.printStackTrace(); 
                } 
            } 
        }
}}).subscribeOn(Schedulers.newThread()).buffer(2, TimeUnit.SECONDS).subscribe(
    new Action1<List<String>>() { 
        @Override public void call(List<String> strings) { 
            for (int i=0;i<strings.size();i++) { 
                Log.d("SampleTransformingActiv", "strings:" + strings.get(i)); 
            } 
            Log.d("SampleTransformingActiv", "==========="); 
        }
});

运行结果

02-15 17:15:22.935 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email1
02-15 17:15:22.935 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email2
02-15 17:15:22.935 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email3
02-15 17:15:22.935 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email4
02-15 17:15:22.945 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: ===========
02-15 17:15:24.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email5
02-15 17:15:24.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email6
02-15 17:15:24.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email7
02-15 17:15:24.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email8
02-15 17:15:24.935 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: ===========
02-15 17:15:26.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email9
02-15 17:15:26.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: strings:email10
02-15 17:15:26.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: ===========
02-15 17:15:28.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: ===========
02-15 17:15:30.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: ===========
02-15 17:15:32.925 14127-14156/com.renyu.rxdemo D/SampleTransformingActiv: ===========

注意观察时间，这边就是每2秒去执行一次操作

FlatMap操作符

Observable.just("1", "2", "3", "4", "5", "6").flatMap(
    new Func1<String, Observable<String>>() { 
        @Override public Observable<String> call(String s) { 
            return Observable.just(s+"2"); 
        }
    }).subscribe(new Action1<String>() { 
        @Override public void call(String s) { 
            Log.d("SampleTransformingActiv", s); 
        }
});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: 12
com.renyu.rxdemo D/SampleTransformingActiv: 22
com.renyu.rxdemo D/SampleTransformingActiv: 32
com.renyu.rxdemo D/SampleTransformingActiv: 42
com.renyu.rxdemo D/SampleTransformingActiv: 52
com.renyu.rxdemo D/SampleTransformingActiv: 62

flatMap就是根据你的规则，将Observable转换之后再发射出去，注意最后的顺序很可能是错乱的，如果要保证顺序的一致性，要使用concatMap

SwitchMap操作符

Observable.just(10, 20, 30).switchMap(
    new Func1<Integer, Observable<Integer>>() { 
        @Override public Observable<Integer> call(Integer integer) { 
            //10的延迟执行时间为1000毫秒、20和30的延迟执行时间为50毫秒 
            int delay = 1000; 
            if (integer>10) { delay=50; } 
            return Observable.just(integer).delay(delay, TimeUnit.MILLISECONDS); 
        }
    }).subscribe(new Action1<Integer>() { 
        @Override public void call(Integer integer) { 
            Log.d("SampleTransformingActiv", "integer:" + integer); 
        }
});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: integer:30

switchmap的功能与flatMap应该说是类似，都是转换Observable，但是区别在于switchmap在产生新的Observable的同时，会摒弃旧的Observable，所以本例并不像之前flatmap一样会产生3个结果，而是直接最后一个Observable替代了之前2个

groupBy操作符

Observable.just("1", "2", "3", "4", "5", "6").groupBy(
    new Func1<String, Boolean>() { 
        @Override public Boolean call(String s) { 
            return Integer.parseInt(s)>3; 
        }
}).subscribe(new Action1<GroupedObservable<Boolean, String>>() { 
      @Override public void call(final GroupedObservable<Boolean, String> booleanStringGroupedObservable) {  
           booleanStringGroupedObservable.subscribe(new Action1<String>() { 
           @Override public void call(String s) { 
               Log.d("SampleTransformingActiv", booleanStringGroupedObservable.getKey() + " " + s); 
            } 
       }); 
}});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: false 1
com.renyu.rxdemo D/SampleTransformingActiv: false 2
com.renyu.rxdemo D/SampleTransformingActiv: false 3
com.renyu.rxdemo D/SampleTransformingActiv: true 4
com.renyu.rxdemo D/SampleTransformingActiv: true 5
com.renyu.rxdemo D/SampleTransformingActiv: true 6

groupBy是通过key建立规则，通过这个规则进行分组发射

Map操作符

Observable.just("1", "2", "3", "4", "5", "6").map(
    new Func1<String, Integer>() { 
        @Override public Integer call(String s) { 
            return Integer.parseInt(s); 
        }
    }).subscribe(new Action1<Integer>() { 
        @Override public void call(Integer integer) { 
            Log.d("SampleTransformingActiv", "integer:" + integer); 
        }
});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: integer:1
com.renyu.rxdemo D/SampleTransformingActiv: integer:2
com.renyu.rxdemo D/SampleTransformingActiv: integer:3
com.renyu.rxdemo D/SampleTransformingActiv: integer:4
com.renyu.rxdemo D/SampleTransformingActiv: integer:5
com.renyu.rxdemo D/SampleTransformingActiv: integer:6

map与flatmap在功能上是一致的，区别在于flatmap是通过中间Observable来进行，而map是直接执行

Cast操作符

Observable.just("1", "2", "3", "4", "5", "6").cast(String.class).subscribe(
    new Action1<String>() { @Override public void call(String integer) { 
        Log.d("SampleTransformingActiv", "integer:" + integer); 
    }
});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: integer:1
com.renyu.rxdemo D/SampleTransformingActiv: integer:2
com.renyu.rxdemo D/SampleTransformingActiv: integer:3
com.renyu.rxdemo D/SampleTransformingActiv: integer:4
com.renyu.rxdemo D/SampleTransformingActiv: integer:5
com.renyu.rxdemo D/SampleTransformingActiv: integer:6

cast用于直接强制转换

Scan

Observable.just(1, 2, 3, 4, 5, 6).scan(new Func2<Integer, Integer, Integer>() { 
    @Override public Integer call(Integer integer, Integer integer2) { 
        return integer+integer2; 
    }
}).subscribe(new Action1<Integer>() { 
    @Override public void call(Integer integer) { 
        Log.d("SampleTransformingActiv", "integer:" + integer); 
    }
});

运行结果

com.renyu.rxdemo D/SampleTransformingActiv: integer:1
com.renyu.rxdemo D/SampleTransformingActiv: integer:3
com.renyu.rxdemo D/SampleTransformingActiv: integer:6
com.renyu.rxdemo D/SampleTransformingActiv: integer:10
com.renyu.rxdemo D/SampleTransformingActiv: integer:15
com.renyu.rxdemo D/SampleTransformingActiv: integer:21

scan类似递归操作，将这个函数的结果作为下一个数据的第一个参数，每次计算结果都会把结果给订阅者

Window
window的作用与Buffer雷同，只不过他是返回一个Observable，而Buffer返回是一个list，这里就不再多说了

主要参考文章

给 Android 开发者的 RxJava 详解扔物线大神的文章，我的RXJava启蒙文章
木水川的博客对RXJava操作符做了很详细的说明以及提供详细的示例
**ReactiveX文档中文翻译 mcxiaoke翻译的RXJava中文文档 **
Android RxJava使用介绍 job_hesc对RXJava操作符介绍以及简单的示例

    原文作者：皮球二二
    原文地址: https://www.jianshu.com/p/4b87b6a86ba0
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。