流

Java的流

如果我们仔细思考，流并不是一个新概念。在Java中，一个Collection可以被看作是一个流，其中Collection中的每个元素都是流中发出的一项数据。

在Android开发中，点击事件可以被视为一个流，位置更新可以是一个流，推送通知也可以是一个流，等等。

命令式与声明式方法

传统上，Java中处理数据流是通过命令式方式完成的。例如，给定一个User对象列表，假设我们只想返回那些拥有博客的用户。这个函数可能如下所示：

1/**
2 * 返回所有拥有博客的用户。
3 * @param users 用户列表
4 * @return 拥有博客的用户
5 */
6public static List<User> getUsersWithABlog(List<User> users) {
7    List<User> filteredUsers = new ArrayList<>();
8    for (User user : users) {
9        if (user.hasBlog()) {
10            filteredUsers.add(user);
11        }
12    }
13    return filteredUsers;
14}

上述代码看起来非常熟悉：一个循环遍历集合中的每个元素，使用if语句判断条件是否成立，对于符合条件的项执行相应操作，即添加到返回列表中。

Java 8 流

由于处理数据流非常普遍，Java 8引入了java.util.stream包。本质上，Java 8流的目标是提高处理流时的抽象层次，提供一种声明式的方式来替代传统的命令式方式。

在上面展示的命令式方法中，我们明确指定了如何处理流；而采用声明式方法时，我们只需说明想要对流做什么。这种看待流处理的方式在函数式编程语言中十分常见，现在也作为核心语言的一部分引入了若干函数式编程特性。使用流后，上述代码可以简化为：

1/**
2 * 返回所有拥有博客的用户。
3 * @param users 用户列表
4 * @return 拥有博客的用户列表
5 */
6public static List<User> getUsersWithABlog(List<User> users) {
7    return users.stream()
8                .filter(user -> user.hasBlog())
9                .collect(Collectors.toList());
10}

使用流，同样的操作可以更加简洁地实现。首先将List转换为Stream，然后过滤出拥有博客的用户，最后再将Stream转换回List。

RxJava 的流

在RxJava的世界里，一切都可以被建模为流。流会随时间发出一项或多项数据，每一项数据在传递过程中都可以被转换或修改。观察者或消费者可以订阅该流，并对每次发出的数据执行相应操作。此外，RxJava中的流具有高度可组合性，甚至可以与其他流结合以产生所需的结果。

为什么要使用RxJava流？

现在我们可能会问：既然RxJava中一切都是流，而且Java 8也支持流，那为什么不用Java 8的流呢？

• RxJava能够让我们在高层实现真正的响应式编程，而Java 8的流做不到这一点。
• 虽然Java 8和RxJava都共享诸如map、filter、merge等函数式语言构造工具箱，但RxJava的流更易于使用。尤其在处理异步操作时，其优势更加明显。

示例

让我们通过一个例子来了解如何使用RxJava解决一个问题。

问题

以我们的示例为例，假设user.hasBlog()是一个网络请求操作，它会阻塞当前线程直到收到响应。这在生产环境中通常不是理想的做法，但为了教学目的，我们暂且接受这一设定。在这种情况下，无论是命令式方法还是Java 8流的方式，在调用.getUsersWithABlog(List<User>)时都会阻塞调用线程。因此，该方法不能从UI线程调用，必须采用某种方式将其移至后台线程执行。

解决方案

使用RxJava，这个问题的解决方案变得非常简单。

1/**
2 * 返回所有拥有博客的用户。
3 * @param users 用户列表
4 * @return 一个发出拥有博客用户的Observable
5 */
6public static Observable<User> getUsersWithABlog(List<User> users) {
7    return Observable.fromIterable(users)
8            .filter(user -> user.hasBlog())
9            .subscribeOn(Schedulers.io());
10}

这里出现了一些新类，我们稍后会深入探讨。但请注意，尽管操作现在运行在后台线程上（通过.subscribeOn(Schedulers.io())），代码的可读性并未受到显著影响。本质上，我们已经声明性地指定了操作应在哪个线程上执行。不过，我们也修改了函数的签名，改为返回Observable<User>。正如我们将看到的，这是迈向Android上响应式编程的第一步。

点击“运行”按钮！

我们在平台上为你准备了一个Gradle Java应用，用于即时实践RxJava的概念。

点击下方代码的“运行”按钮，亲身体验上述示例的实际效果！