Kotlin，RxJava与RxAndroid混合开发

通过前几篇Kotlin文章 ：

今天到第5篇，谈谈Kotlin和RxJava。

Kotlin，RxJava这是一个令人难以相信强大的组合，我们将利用两者的优势，为我们的应用程序提供一个新的功能，并将我们的主UI线程与我们的后台任务解耦，在这种情况下，请求webd的Reddit News。

本次我打算放弃所有一切准备，不打算启用一个与真正的服务器（下一部分）连接，但现在模拟数据将是可以模拟一个服务器请求。

如果这是你第一次听说RxJava，我鼓励你使用它，阅读这个令人难以置信的软件和平，这将改变你的思想。

RxJava是Reactive Extensions的实现，Reactive Extensions是一个通过使用可观察序列来编写异步和基于事件的程序的库。

我们将使用Observable类来避免从的Main UI Thread中发出服务器请求（或者任何消耗长时间的执行）。

NewsManager

NewsManager类负责提供来自Reddit API的新闻。

它将负责执行服务器请求，并使用我们已经创建的新闻UI模型给我们一个新闻列表。（https://github.com/juanchosaravia/KedditBySteps/blob/master/app/src/main/java/com/droidcba/kedditbysteps/commons/Models.kt）。

主要思想是使我们的API调用在另一个线程中被执行，并且不在主UI线程中。

一张图解释是这样的：

可观察的Observable

NewsManager类将提供一个方法，该方法将返回一个Observable对象

，它将允许您在另一个上下文（在本例中是在一个新线程中）运行一段代码（如API调用）。

fun getNews(): Observable { ...}

这个Observable对象将由我们从NewsManager创建。

我们将提供这个对象的实现：

fun getNews(): Observable { return Observable.create { subscriber -> val news = mutableListOf() // api call... subscriber.onNext(news) subscriber.onCompleted() }}

强大的Kotlin，我们可以使用lambda表达式来提供Observable.create（...）

方法所需的函数，在这种情况下，我们将在Observable提供的上下文中执行的代码的实现（我们在后面会看到更多关于这个上下文的内容，但总之这意味着在哪个线程中执行）。

create（）方法要求你提供一个接收订阅对象的函数，这个订阅对象允许你通过执行3种可能的方法发送事件给订阅者：

onNext（item：T）：我们打算调用这个方法来向订阅者提供从API调用收到的消息。

在这种情况下，T将是我们的“List ”，它已经从getNews（）签名的返回类型推断出来了。

onCompleted（）：当我们完成提供来自这个Observable的所有数据时，调用这个方法，在这种情况下，我们将在从onNext（）方法发送消息之后调用它，并且我们准备完成这个过程。

onError（e：Throwable）：如果我们面对API调用的任何问题，我们将调用这个方法告诉订阅者有错误。

稍后我将使用它通过显示SnackBar消息来通知用户请求出现的问题。

提交

在这里，你可以获得添加NewsManager和getNews（）方法定义所需的所有代码：https://github.com/juanchosaravia/KedditBySteps/commit/1623947fa709ba156d41d7a67f016871c8e9bd0d

所以我们通过提供模拟的消息来模拟我们正在请求消息，并调用onNext（）和onCompleted（）来通知所有订阅者这个信息（在另一个提交中添加了onCompleted方法）

NewsManager

在我们的NewsManager中添加了我们在本教程的第三部分（Kotlin 3：NewsFragment的扩展，Android相关类扩展）中讨论的，

懒惰属性：

private val newsManager by lazy { NewsManager() }

这个NewsManager只会在我们使用newsManagers字段的时候才用

NewsManager（）进行初始化。

NewsFragment Request News

现在是时候使用我们的NewsManager，并要求我们的RecyclerView显示一些消息。

订阅>订阅>观察者

为了NewsManager 的Reddit消息，我们将把NewsFragment转换为NewsManager.getNews（）获得的Observable的Observer。

我们将通过调用getNews（），方法来执行此操作，并从接收到的Observable中调用“subscribe（...）” 方法：

val subscription = newsManager.getNews().subscribe ( { retrievedNews -> ... }, { e -> ... }

订阅的方法

有几个重载，我们打算使用这个：

public final Subscription subscribe( final Action1

它接收两个功能：

onNext：当Observable调用我们之前看到的onNext（）方法时被调用的函数。

我们将用它来设置NewsAdapter与收到的消息。

onError：调用Observable的onError（）方法时调用的函数。

我们将用它来显示一个带有错误信息的SnackBar。

RxJava返回一个Subscription。

这个对象允许我们管理订阅，比如检查订阅是否取消订阅（我们将在本教程的最后看到更多的内容）。

再次感谢Kotlin，我们可以使用lambda表达式提供这两个函数。

例如在第一个函数中，retrieveNews 变量 ，是我给从onNext（）方法收到的消息的名称：

{ retrievedNews -> (news_list.adapter as NewsAdapter).addNews(retrievedNews)}

请记住，onNext返回类型是“List ”，所以我所做的就是直接将这些消息传递给我们的NewsAdapter。

而对于onError函数，我只是告诉用户SnackBar有错误。

我可以使用“e”（这是一个Throwable类型）来获取有关接收到的异常的更多详细信息：

我们仍然处于主线

如果我们运行该应用程序，它将会正常工作，因为我们正在消费这个Observable提供的数据。

但是如果你真的做了一个长时间运行的操作，而不是我们生成的模拟数据，那么应用程序将停止工作，因为我们的操作仍然处在主UI线程中。

因为我们没有提供任何具体的细节给我们的Observable，

所以它将以默认的配置运行，这个配置将在调用它的同一个线程中执行。

因此，让我们将Observable配置为在另一个线程中执行，但仍然需要通知当前主UI线程中的消费任何事件。

SubscribeOn

val subscription = newsManager.getNews（）. subscribeOn（Schedulers.io（））. subscribe（...）

subscribeOn（...）方法允许您将Observable代码的执行切换到另一个线程，并具有特定的行为。

为此，RxJava使用调度程序，它提供了你所需的行为的用例。

RxJava还提供了常见场景的调度程序列表：

io：意图用于IO界的工作。

compare：用于计算工作。

newThread：为每个工作单元创建一个新的Thread。

test：用于调试。

在我们的例子中，我们将使用Schedulers.io（），

因为我们将要执行一个API请求。

ObserveOn

ObserveOn是Observable类中的另一种方法，它允许您定义要执行“subscribe（...）” 方法中提供的函数的位置。

换句话说，在哪个线程将要执行我们在订阅方法中用onNext和onError代码提供的函数。

因为我们需要在主UI线程中执行这个代码，所以使我们的Observable可以观察到主UI线程是一个好的方式。

这里是我们需要添加RxAndroid的依赖关系来调用一个新的调度器：

AndroidSchedulers.mainThread（）

需要更新我们的代码来设置observeOn Scheduler：

val subscription = newsManager.getNews（）. subscribeOn（Schedulers.io（））. observeOn（AndroidSchedulers.mainThread（））. subscribe（... ）

提交

https://github.com/juanchosaravia/KedditBySteps/commit/c254d86b03fdbef24ac847445d2e7a6034201330

添加 RxAndroid

https://github.com/juanchosaravia/KedditBySteps/commit/041903f61674a06be84fa19429edbef9a5aaa5f8

随着最后的变化，我们正在另一个子线程中创建我们的模拟数据，并在屏幕上显示模拟的新闻:

但我们必须意识到，我们的应用程序可以随时关闭，或者在订阅中间进行方向更改操作。

如果我们不正确地管理我们的订阅，这可能会导致我们遇到问题，所以我们来做一些事情吧。

订阅

为了避免在我们从一个片段离开或关闭我们的应用程序的过程中离开我们的订阅，我们必须仔细管理它，一个好的做法是取消订阅我们创建的所有订阅。

一个很好的方法是将所有您创建的订阅添加到

CompositeSubscription对象中，该类由RxJava提供，并允许您仅使用一个方法调用就可以取消订阅所有订阅。

我们将在onResume()方法中初始化一个CompositeSubscription对象，并在onPause方法被调用时取消订阅。

代码将如下所示：

var subscriptions = CompositeSubscription()override fun onResume() { super.onResume() subscriptions = CompositeSubscription()}override fun onPause() { super.onPause() subscriptions.clear()}

我们现在唯一要做的就是把我们的订阅添加到这个新的CompositeSubscription对象中，我们可以随时离开我们的应用程序（或片段）：

private fun requestNews() { val subscription = newsManager.getNews() .subscribeOn(Schedulers.io()) .subscribe (...)

subscriptions.add(subscription) // add the subscription}

为了使它更具可读性，我将所有这些逻辑移动到一个新的名为“RxBaseFragment”的基本片段中，默认情况下给出了所有这些逻辑。

结账的代码看到更多的细节。

https://github.com/juanchosaravia/KedditBySteps/commit/ff2fee38a7e546bc8ab7d47db2cd4c6bd49eeb80

结论

正如你所看到的，Kotlin＆RxJava完美地结合在一起，在本教程中，我们只看到了RxJava中强大功能中一小部分，Kotlin的代码完全可读，并且易于理解。