1.变换操作

scan( ) — 对Observable发射的每一项数据应用一个函数，然后按顺序依次发射每一个值

Observable.just(1,2,3)        .scan(newFunc2() {@OverridepublicIntegercall(Integer integer, Integer integer2){//自定义函数returninteger + integer2;            }        })        .subscribe(newAction1() {@Overridepublicvoidcall(Integer integer){                Log.d("RxJava", String.valueOf(integer));            }        });

结果：将自定义函数应用于数据序列，并将这个函数的结果作为函数下一次的参数1使用，1+0=1，1+2=3 ，3+3=6

11-06 02:54:43.040 25433-25433/keye.com.rxjavaobserver D/RxJava: 111-06 02:54:43.040 25433-25433/keye.com.rxjavaobserver D/RxJava: 311-06 02:54:43.040 25433-25433/keye.com.rxjavaobserver D/RxJava: 6

groupBy( ) — 将Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据

Observable.just(1,2,3,4)        .groupBy(newFunc1() {@OverridepublicIntegercall(Integer integer){//这里返回的结果为keyreturninteger +1;            }        })        .subscribe(newAction1>() {@Overridepublicvoidcall(GroupedObservable groupedObservable){                groupedObservable.subscribe(newAction1() {@Overridepublicvoidcall(Integer integer){                        Log.d("RxJava","key:"+ groupedObservable.getKey() +",value:"+ integer);                    }                });            }        });

结果：在第一个func1函数中，设置key，最后生成一个Observable集合，并把每一个groupedObservable，并依次发射出去

3162-3162/keye.com.rxjavaobserver D/RxJava: key:2,value:13162-3162/keye.com.rxjavaobserver D/RxJava: key:3,value:23162-3162/keye.com.rxjavaobserver D/RxJava: key:4,value:33162-3162/keye.com.rxjavaobserver D/RxJava: key:5,value:4

buffer( ) — 它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个

Observable.just(1,2,3,4,5,6,7)          .buffer(2,1)          .subscribe(newAction1>() {@Overridepublicvoidcall(List integers){                Log.d("RxJava", integers +"");              }});

结果：buffer()有两个参数count和skip，count指定List的大小，skip指定每次发射一个List需要跳过几个数；buffer(2, 1)：每组2个数，每次跳过1个数，结果如下：

keye.com.rxjavaobserver D/RxJava: [1, 2]keye.com.rxjavaobserver D/RxJava: [4, 5]keye.com.rxjavaobserver D/RxJava: [7]

window( ) — 定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项

Observable.just(1,2,3,4,5,6,7)          .window(2,2)          .subscribe(newAction1>() {@Overridepublicvoidcall(Observable observable){                  Log.d("RxJava","window");                  observable.subscribe(newAction1() {@Overridepublicvoidcall(Integer integer){                          Log.d("RxJava", integer +"");                      }                  });              }});

结果：window()操作符和buffer()类似，都是缓存一段数据集合，再打包发射出去

11-0603:30:29.19825883-25883/? D/RxJava:window11-0603:30:29.19825883-25883/? D/RxJava:111-0603:30:29.19825883-25883/? D/RxJava:211-0603:30:29.19825883-25883/? D/RxJava:window11-0603:30:29.19825883-25883/? D/RxJava:311-0603:30:29.19825883-25883/? D/RxJava:411-0603:30:29.19825883-25883/? D/RxJava:window11-0603:30:29.19825883-25883/? D/RxJava:511-0603:30:29.19925883-25883/? D/RxJava:611-0603:30:29.19925883-25883/? D/RxJava:window11-0603:30:29.19925883-25883/? D/RxJava:7