运用RxJava（二）操作符

在本系列文章的第一部分我讲述了RxJava的基本结构，也向你介绍了map()操作符。但是，我能够理解你还并不能够使用它——你所掌握的还没有到能够使用它的程度。这种状况即将改变——RxJava大部分的能力都来自于这个框架中的操作符。

举个例子来介绍它的操作符吧。

栗子

假定我现在有这样一个方法：

// Returns a List of website URLs based on a text search
Observable<List<String>> query(String text);

我想开发一个能够搜索文本并展示结果的系统，根据上一篇文章我们所学到的，可以写出如下代码：

query("Hello, world!")
    .subscribe(urls -> {
        for (String url : urls) {
            System.out.println(url);
        }
    });

这个答案自然恨不能让人满足，因为我就此丧失了变换数据流的能力。如果我想修改每一个 url，我就必须在Subscriber中完成这一切。我们也就此抛弃了使用map()的技巧。

希望之光

有一个Observable.from()的方法，能够将传入的集合拆分成具体的对象依次发送出来：

Observable.from("url1", "url2", "url3")
    .subscribe(url -> System.out.println(url));

看起来像是有用的，那就看看发生了什么吧。

query("Hello, world!")
    .subscribe(urls -> {
        Observable.from(urls)
            .subscribe(url -> System.out.println(url));
    });

我拿到了一个foreach循环，可结果代码却是一团糟。我现在到手的是多重的嵌入的订阅，除了丑陋、难以修改外，它还破坏了RxJava的一些重要的（尽管我们还不知晓）特征。我去！

更好的方法

屏气凝神迎接你的救星吧：flatMap()。

Observable.flatMap()传入一个Observable发送的数据，返回另一个Observable发送的数据。突如其来的变化：你以为你得到的是一个数据流其实却是另外一个。看看它是怎么解决这个问题的：

query("Hello, world!")
    .flatMap(new Func1<List<String>, Observable<String>>() {
        @Override
        public Observable<String> call(List<String> urls) {
            return Observable.from(urls);
        }
    })
    .subscribe(url -> System.out.println(url));

上面是完整的函数，这样你就能确切地知道到底发送了什么，但如果用lambdas来简化代码的话，看起来就很清爽了：

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .subscribe(url -> System.out.println(url));

flatMap()有点奇怪吧？它为什么会返回另外一个Observable呢？这里的关键点是Subscriber看到的就是返回的新的Observable。它并没有收到一个List<String>——它收到的是Observable.from()返回的一系列单独的String。

还可以更好

我必须着重强调：flatMap()能够返回任意它想返回的Observable。

假定现在我又有一个方法：

// Returns the title of a website, or null if 404
Observable<String> getTitle(String URL);

现在我想打印得到的每一个网站的标题而不再是URL，但有几个问题：我的方法同时只能作用在单独的URL上，并且它并不返回一个String，而是返回一个发送着String的Observable。

使用flatMap()来解决这个问题其实很简单，将URL列表分割完成之后，在URL到达Subscriber之前，我可以在flatMap()中对每个URL使用getTitle()：

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(new Func1<String, Observable<String>>() {
        @Override
        public Observable<String> call(String url) {
            return getTitle(url);
        }
    })
    .subscribe(title -> System.out.println(title));

再一次使用lambdas简化代码：

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .subscribe(title -> System.out.println(title));

牛吧？我将多个单独返回Observable的方法组合在了一起。爽！

还不止如此，注意我是怎么把两个API调用合并到一个链式调用中的。这可以扩展到任意多个API调用。你知道保证API调用同步有多痛苦吗？但我们已经跳过了回调的地狱，所有这些的相同逻辑都被封装在了这个简短的reactive调用当中。

丰富的操作符

到现在为止我们也就学了两个操作符，但其实它还有很多。那么，我们还能怎么优化我们的代码呢？

getTitle()将返回null如果URL遭遇404错误。但我们不想输出“null”，事实证明我们可以将它过滤掉：

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .subscribe(title -> System.out.println(title));

filter()原样发送收到的通过boolean检查的数据。

现在我们一次最多只想显示5条数据：

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .take(5)
    .subscribe(title -> System.out.println(title));

take()一次最多发送指定的数据数目（如果少于5条就提前结束）。

现在，我们希望将每个标题存盘：

query("Hello, world!")
    .flatMap(urls -> Observable.from(urls))
    .flatMap(url -> getTitle(url))
    .filter(title -> title != null)
    .take(5)
    .doOnNext(title -> saveTitle(title))
    .subscribe(title -> System.out.println(title));

doOnNext()允许我们在数据被发送之前添加额外的行为，这个例子中就是将标题存盘。

瞧瞧操作数据流有多么简单吧。你可以给你的程序添加更多技巧但不会引起混乱。

RxJava有着丰富的操作符，这么多的操作符的确有些吓人但值得复习一下，这样你就知道哪些有用。消化这些操作符需要一些时间，但是一旦你这么做了，你手上就有了洪荒之力。

根据上面提供的那些，你甚至还可以自定义自己的操作符。那超出了本文的范围，但只要你想做，你一定可以做到的。

然后呢？

好吧，看来你是一个很难满足的怀疑论者。为什么你应该关注所有这些操作符？

关键概念 #3: 操作符能够让你对数据流做任何事！

唯一的限制就是你自己

你能够只使用简单的操作符链就建立起复杂的逻辑。它能够将你的代码分解成可组合的片段。那就是功能性的reactive编程。你使用得越多，你思考程序的方式改变得就会越多。

另外，试想一下，要去消费我们变换完成后的数据是多么的简单。在这个例子的最后我们使用了两个API调用，操作了数据然后存盘，但我们的Subscriber却并不知道那些，它只会认为它消费了一个简单的Observable<String>。封装，使得编程更为简易！

在本文的第三部分，我们将学到RxJava中不直接涉及操作数据的一些其他特征。比如错误处理和并发。

运用RxJava（二）操作符