RxJava常用操作符

创建
- unfaseCreate(create)
  创建一个Observable（被观察者），当被观察者（Observer）/订阅者（Subscriber）
  subscribe(订阅)的时候就会依次发出三条字符串数据（"Hello","RxJava","Nice to meet you"）
  最终onComplete

Observable.unsafeCreate(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello");
                subscriber.onNext("RxJava");
                subscriber.onNext("Nice to meet you");
                subscriber.onCompleted();
            }
        });

just
作用同上，订阅时依次发出三条数据，不过此方法参数可以有1-9条

Observable.just("Hello", "RxJava", "Nice to meet you")

from
作用同just不过是把参数封装成数组或者可迭代的集合在依次发送出来，突破了just9个参数的限制

String[] strings = {"Hello", "RxJava", "Nice to meet you"};
Observable.from(strings)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

defer
顾名思义，延迟创建

private String[] strings1 = {"Hello", "World"};
    private String[] strings2 = {"Hello", "RxJava"};

    private void test() {
        Observable<String> observable = Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return Observable.from(strings1);
            }
        });

        strings1 = strings2; //订阅前把strings给改了
        observable.subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

我们发现数据结果变成这样了

onNext--> Hello
onNext--> RxJava
onComplete

由此可以证明defer操作符起到的不过是一个“预创建”的作用，真正创建是发生在订阅的时候

empty
创建一个空的，不会发射任何事件（数据）的Observable

Observable.empty()
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onComplete

never
创建一个不会发出任何事件也不会结束的Observable

Observable.never()
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

······

error
创建一个会发出一个error事件的Observable

Observable.error(new RuntimeException("fuck!"))
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onError--> fuck!

range
创建一个发射一组整数序列的Observable

Observable.range(3, 8)
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 3
onNext--> 4
onNext--> 5
onNext--> 6
onNext--> 7
onNext--> 8
onNext--> 9
onNext--> 10
onComplete

interval
创建一个无限的计时序列，每隔一段时间发射一个数字（从0开始）的Observable

   Observable.interval(1, TimeUnit.SECONDS)
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

   System.in.read();//阻塞当前线程，防止JVM结束程序

onNext--> 0
onNext--> 1
onNext--> 2
onNext--> 3
onNext--> 4
onNext--> 5
onNext--> 6
...

转换
- buffer(int count)
  将原发射出来的数据已count为单元打包之后在分别发射出来

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        .buffer(3)
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> [1, 2, 3]
onNext--> [4, 5, 6]
onNext--> [7, 8, 9]
onNext--> [10]
onComplete

map
将原Observable发射出来的数据转换为另外一种类型的数据

Observable.just("Hello", "RxJava", "Nice to meet you")
        .map(new Func1<String, Integer>() { //泛型第一个类型是原数据类型，第二个类型是想要变换的数据类型
            @Override
            public Integer call(String s) {
                // 这是转换成了Student类型
                // Student student = new Student();
                // student.setName(s);
                // return student;
                return s.hashCode();        //将数据转换为了int（取得其hashCode值）
            }
        })
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 69609650
onNext--> -1834252888
onNext--> -1230747480
onComplete

flatMap
作用类似于map又比map强大，map是单纯的数据类型的转换，而flapMap可以将原数据新的Observables，再将这些Observables的数据顺序缓存到一个新的队列中，在统一发射出来

Observable.just("Hello", "RxJava", "Nice to meet you")
        .flatMap(new Func1<String, Observable<Integer>>() {
            @Override
            public Observable<Integer> call(String s) {
                return Observable.just(s.hashCode());
            }
        })
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer o) {
                System.out.println("onNext--> " + o);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

onNext--> 69609650
onNext--> -1834252888
onNext--> -1230747480
onComplete

这里虽然结果和map是相同的，但是过程却是不同的。flatMap是先将原来的三个字符串（"Hello","RxJava","Nice to meet you"）依次取其hashCode，在利用Observable.just将转换之后的int类型的值在发射出来。map只是单穿的转换了数据类型，而flapMap是转换成了新的Observable了，这在开发过程中遇到嵌套网络请求的时候十分方便。

window
看到网上其他人说他的总用类似于buffer，不过我倒是认为他更像flatMap，区别的是flapMap在转换之后形成新的Observable会再将新的数据发射出来，不过window就仅仅只转换成了发射新的数据类型的Observable，有点像是flatMap在干活时半途而废的意思。
过滤
- filter
  对发射的数据做一个限制，只有满足条件的数据才会被发射

Observable.just("Hello", "RxJava", "Nice to meet you")
        .filter(new Func1<String, Boolean>() {
            @Override
            public Boolean call(String s) {
                //这里的显示条件是s的长度大于5，而Hello的长度刚好是5
                //所以不能满足条件
                return s.length() > 5;
            }
        })
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> RxJava
onNext--> Nice to meet you
onComplete

take
只发射前N项的数据（takeLast与take想反，只取最后N项数据）

Observable.just("Hello", "RxJava", "Nice to meet you")
        .take(2)
        //.taktLast(2)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

onNext--> Hello
onNext--> RxJava
onComplete

skip
发射数据时忽略前N项数据（skpiLast忽略后N项数据）

Observable.just("Hello", "RxJava", "Nice to meet you")
        .skip(2)
        //.skipLast(2)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
       }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });
    }

onNext--> Nice to meet you
onComplete

elementAt
获取原数据的第N项数据作为唯一的数据发射出去（elementAtOrDefault会在index超出范围时，给出一个默认值发射出来）

Observable.just("Hello", "RxJava", "Nice to meet you")
        .elementAtOrDefault(1, "Great")
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> RxJava
onComplete

distinct
过滤掉重复项

Observable.just("Hello", "Hello", "Hello", "RxJava", "Nice to meet you")
        .distinct()
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> Hello
onNext--> RxJava
onNext--> Nice to meet you
onComplete

组合
- startWith
  在发射一组数据之前先发射另一组数据

Observable.just("Hello", "RxJava", "Nice to meet you")
        .startWith("One", "Two", "Three")
        .subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> One
onNext--> Two
onNext--> Three
onNext--> Hello
onNext--> RxJava
onNext--> Nice to meet you
onComplete

merge
将多个Observables发射的数据合并后在发射

Observable.merge(Observable.just(1, 2, 3), Observable.just(4, 5),
        Observable.just(6, 7), Observable.just(8, 9, 10))
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 1
onNext--> 2
onNext--> 3
onNext--> 4
onNext--> 5
onNext--> 6
onNext--> 7
onNext--> 8
onNext--> 9
onNext--> 10
onComplete

zip
按照自己的规则发射与发射数据项最少的相同的数据

Observable.zip(Observable.just(1, 8, 7), Observable.just(2, 5),
        Observable.just(3, 6), Observable.just(4, 9, 0), new Func4<Integer, Integer, Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2, Integer integer3, Integer integer4) {
                return integer < integer2 ? integer3 : integer4;
            }
        })
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer s) {
                System.out.println("onNext--> " + s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("onError--> " + throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                System.out.println("onComplete");
            }
        });

onNext--> 3
onNext--> 9
onComplete

通过观察以上例子可以发现我们的发射规则是如果发射的第一个数据小于第二个数则发射第三个数据，否则发射第四个数据（我们来验证一下，1确实是小于2的，随意发射的是3；8并不小于5，所以发射的是9，又因为四个发射箱，最少的之后两项，所以最后只发射了两项数据）

最后编辑于：2017.12.06 06:27:43

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 194,390评论 5赞 459
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 81,821评论 2赞 371
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 141,632评论 0赞 319
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 52,170评论 1赞 263
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 61,033评论 4赞 355
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 46,098评论 1赞 272
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 36,511评论 3赞 381
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 35,204评论 0赞 253
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 39,479评论 1赞 290
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 34,572评论 2赞 309
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 36,341评论 1赞 326
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 32,213评论 3赞 312
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 37,576评论 3赞 298
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 28,893评论 0赞 17
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 30,171评论 1赞 250
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 41,486评论 2赞 341
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 40,676评论 2赞 335

RxJava常用操作符

创建

转换

过滤

组合

推荐阅读更多精彩内容