我为什么要使用Kotlin协同程序？
似乎RxKotlin库更加多样化.Kotlin协同程序看起来效率显着降低,相比之下使用起来更加麻烦.
我根据安德烈·布雷斯拉夫(JetBrains)的设计讲话对协同程序提出了自己的看法:https://www.youtube.com/watch？v = 4W3ruTWUhpw
可以在此处访问来自谈话的幻灯片:https://www.slideshare.net/abreslav/jvmls-2016-coroutines-in-kotlin
编辑(感谢@hotkey):
关于当前协程状态的更好来源:https://github.com/Kotlin/KEEP/blob/master/proposals/coroutines.md
Geo*_*izy 82
Rx有两个部分; Observable模式,以及一组操作,转换和组合它们的实体操作符.Observable模式本身并没有做太多.与Coroutines相同; 这只是处理异步问题的另一种范式.您可以比较回调,Observable和协同程序的优缺点来解决给定的问题,但是您无法将范例与功能齐全的库进行比较.这就像将语言与框架进行比较一样.
Kotlin协程如何比RxKotlin更好？还没有使用过coroutines,但它看起来类似于C#中的async/wait.您只需编写顺序代码,一切都像编写同步代码一样简单...除了它异步执行.它更容易掌握.
我为什么要使用kotlin协同程序？我会自己回答.大多数时候我会坚持使用Rx,因为我喜欢事件驱动的架构.但是应该出现我正在编写顺序代码的情况,并且我需要在中间调用异步方法,我将很乐意利用协同程序保持这种方式并避免将所有内容包装在Observable中.
编辑:现在我正在使用协同程序,现在是时候进行更新了.
RxKotlin只是在Kotlin中使用RxJava的语法糖,所以我将在下面谈论RxJava而不是RxKotlin.协同程序是比RxJava更低的杠杆和更一般的概念,它们服务于其他用例.也就是说,有一个用例可以比较RxJava和coroutines(channel),它会以异步方式传递数据.协同程序在这里比RxJava有明显的优势:
subscribeOn()和ObserveOn()令人困惑.每个协同程序都有一个线程上下文并返回到父上下文.对于一个渠道,双方(生产者,消费者)都在他自己的背景下执行.协同程序在线程或线程池修饰方面更直观.yield),prioritize(select),parallelize(multiple producer/ actoron channel)或lock resource(Mutex).它可能与服务器(RxJava最先出现)无关,但在资源有限的环境中,可能需要这种级别的控制.send()通道的另一端是一个暂停功能,在达到通道容量时暂停.这是大自然给出的开箱即用的背压.您也可以offer()进行通道,在这种情况下,调用永远不会挂起,但false如果通道已满,则返回,onBackpressureDrop()从RxJava 有效地再现.或者你可以编写自己的自定义背压逻辑,这对于协同程序来说并不困难,特别是与RxJava相同.还有另一个用例,协同程序闪耀,这将回答你的第二个问题"我为什么要使用Kotlin协同程序？".协同程序是后台线程或AsyncTask(Android)的完美替代品.这很简单launch { someBlockingFunction() }.当然,你可以用RxJava做到这一点太,使用Schedulers和Completable可能.你不会(或很少)使用Observer模式和作为RxJava签名的运算符,暗示这项工作超出了RxJava的范围.RxJava复杂性(这里的无用税)将使您的代码比Coroutine的版本更冗长,更简洁.
可读性至关重要.在这方面,RxJava和协同程序方法有很大不同.协同程序比RxJava更简单.如果你不放心用map(),flatmap()和功能反应式编程一般,协程的操作更容易,涉及到基本的指令:for,if,try/catch...但我个人觉得协同程序的代码更难理解为不平凡的任务.特别是它涉及更多的嵌套和缩进,而RxJava中的操作符链保持一致.功能样式编程使处理更加明确.最重要的是,RxJava可以通过其丰富的(OK,方式太丰富)运算符集合中的一些标准运算符来解决复杂的转换.当您有复杂的数据流需要大量的组合和转换时,RxJava会闪耀.
我希望这些考虑因素可以帮助您根据自己的需求选择合适的工具.
Rom*_*rov 79
Kotlin协程与Rx不同.很难比较它们,因为Kotlin协程是一种瘦的语言特性(只有几个基本概念和一些基本功能来操作它们),而Rx是一个非常重的库,有很多种类的即用型运营商.两者都旨在解决异步编程的问题,但他们的解决方法却截然不同:
Rx具有特定的功能编程风格,几乎可以在任何编程语言中实现,而无需语言本身的支持.当手头的问题很容易分解成一系列标准操作符时,它很有效,否则就不那么好了.
Kotlin协程提供了一种语言功能,允许库编写者实现各种异步编程风格,包括但不限于功能反应式(Rx).使用Kotlin协同程序,您还可以以命令式样式,基于承诺/期货的样式,演员样式等编写异步代码.
将Rx与基于Kotlin协同程序实现的某些特定库进行比较更为合适.
以kotlinx.coroutines库为例.该库提供了一组原语async/await和通道,这些原语通常被烘焙到其他编程语言中.它还支持轻量级的未来演员.您可以通过示例在"kotlinx.coroutines指南"中阅读更多内容.
提供的通道kotlinx.coroutines可以在某些用例中替换或扩充Rx.有一个单独的反应流指南与协同程序,更深入地与Rx相似和不同.
Dan*_*ato 42
我非常了解 RxJava，最近我转向了 Kotlin Coroutines 和 Flow。
RxKotlin 与 RxJava 基本相同，只是添加了一些语法糖，使其在 Kotlin 中编写 RxJava 代码更加舒适/惯用。
RxJava 和 Kotlin 协程之间的“公平”比较应该包括 Flow，我将在这里尝试解释原因。这会有点长，但我会尽量用例子来保持它的简单。
使用 RxJava，你有不同的对象（从版本 2 开始）：
// 0-n events without backpressure management
fun observeEventsA(): Observable<String>
// 0-n events with explicit backpressure management
fun observeEventsB(): Flowable<String>
// exactly 1 event
fun encrypt(original: String): Single<String>
// 0-1 events
fun cached(key: String): Maybe<MyData>
// just completes with no specific results
fun syncPending(): Completable
在 kotlin 协程 + 流中，您不需要很多实体，因为如果您没有事件流，您可以只使用简单的协程（挂起函数）：
// 0-n events, the backpressure is automatically taken care off
fun observeEvents(): Flow<String>
// exactly 1 event
suspend fun encrypt(original: String): String
// 0-1 events
suspend fun cached(key: String): MyData?
// just completes with no specific results
suspend fun syncPending()
奖励：Kotlin Flow / Coroutines 支持null值（RxJava 2 移除了支持）
在 RxJava 中，您有很多运算符（map, filter, flatMap, switchMap, ...），并且对于其中的大多数，每种实体类型（Single.map(), Observable.map(), ...）都有一个版本。
Kotlin Coroutines + Flow不需要那么多运算符，让我们通过一些最常见运算符的示例来了解原因
地图（）
RxJava：
fun getPerson(id: String): Single<Person>
fun observePersons(): Observable<Person>
fun getPersonName(id: String): Single<String> {
return getPerson(id)
.map { it.firstName }
}
fun observePersonsNames(): Observable<String> {
return observePersons()
.map { it.firstName }
}
Kotlin 协程 + 流程
suspend fun getPerson(id: String): Person
fun observePersons(): Flow<Person>
suspend fun getPersonName(id: String): String? {
return getPerson(id).firstName
}
fun observePersonsNames(): Flow<String> {
return observePersons()
.map { it.firstName }
}
对于“单一”案例，您不需要运算符，它与Flow案例非常相似。
平面图()
假设您需要为每个人从数据库（或远程服务）中获取它的保险
RxJava
fun fetchInsurance(insuranceId: String): Single<Insurance>
fun getPersonInsurance(id: String): Single<Insurance> {
return getPerson(id)
.flatMap { person ->
fetchInsurance(person.insuranceId)
}
}
fun obseverPersonsInsurances(): Observable<Insurance> {
return observePersons()
.flatMap { person ->
fetchInsurance(person.insuranceId) // this is a Single
.toObservable() // flatMap expect an Observable
}
}
让我们看看 Kotlin Cooutiens + Flow
suspend fun fetchInsurance(insuranceId: String): Insurance
suspend fun getPersonInsurance(id: String): Insurance {
val person = getPerson(id)
return fetchInsurance(person.insuranceId)
}
fun obseverPersonsInsurances(): Flow<Insurance> {
return observePersons()
.map { person ->
fetchInsurance(person.insuranceId)
}
}
像以前一样，在简单的协程情况下，我们不需要运算符，我们只需编写代码，就像它不是异步的那样，只使用挂起函数。
这Flow不是一个错字，不需要flatMap运算符，我们可以使用map. 原因是 map lambda 是一个挂起函数！我们可以在里面执行挂起代码！！！
为此，我们不需要另一个操作员。
对于更复杂的东西，您可以使用 Flowtransform()运算符。
每个 Flow 操作符都接受一个挂起函数！
所以如果你需要，filter()但你的过滤器需要执行网络调用，你可以！
fun observePersonsWithValidInsurance(): Flow<Person> {
return observerPersons()
.filter { person ->
val insurance = fetchInsurance(person.insuranceId)
insurance.isValid()
}
}
延迟（），startWith（），concatWith（），...
在 RxJava 中，您有许多运算符用于在前后应用延迟或添加项目：
使用 kotlin Flow，您可以简单地：
grabMyFlow()
.onStart {
// delay by 3 seconds before starting
delay(3000L)
// just emitting an item first
emit("First item!")
emit(cachedItem()) // call another suspending function and emit the result
}
.onEach { value ->
// insert a delay of 1 second after a value only on some condition
if (value.length() > 5) {
delay(1000L)
}
}
.onCompletion {
val endingSequence: Flow<String> = grabEndingSequence()
emitAll(endingSequence)
}
错误处理
RxJava 有很多操作符来处理错误：
使用 Flow，您只需要操作员即可catch()：
grabMyFlow()
.catch { error ->
// emit something from the flow
emit("We got an error: $error.message")
// then if we can recover from this error emit it
if (error is RecoverableError) {
// error.recover() here is supposed to return a Flow<> to recover
emitAll(error.recover())
} else {
// re-throw the error if we can't recover (aka = don't catch it)
throw error
}
}
并且具有挂起功能，您可以只使用try {} catch() {}.
易于编写流操作符
由于协程在幕后为 Flow 提供动力，因此编写运算符更容易。如果您曾经检查过 RxJava 操作符，您就会发现它有多难以及需要学习多少东西。
编写 Kotlin Flow 运算符更容易，您只需查看此处已成为 Flow 一部分的运算符的源代码即可获得想法。原因是协程使编写异步代码变得更容易，并且操作符使用起来更自然。
作为奖励，Flow 运算符都是 kotlin 扩展函数，这意味着您或库都可以轻松添加运算符，并且使用起来不会感到奇怪（例如observable.lift()或observable.compose()）。
上游线程不会向下游泄漏
这到底是什么意思？
让我们以这个 RxJava 为例：
urlsToCall()
.switchMap { url ->
if (url.scheme == "local") {
val data = grabFromMemory(url.path)
Flowable.just(data)
} else {
performNetworkCall(url)
.subscribeOn(Subscribers.io())
.toObservable()
}
}
.subscribe {
// in which thread is this call executed?
}
那么回调在哪里subscribe执行呢？
答案是：
要看...
如果它来自网络，它在一个 IO 线程中；如果它来自另一个分支，则它是未定义的，这取决于用于发送 url 的线程。
这就是“上游线程向下游泄漏”的概念。
对于 Flow 和 Coroutines，情况并非如此，除非您明确要求此行为（使用Dispatchers.Unconfined）。
suspend fun myFunction() {
// execute this coroutine body in the main thread
withContext(Dispatchers.Main) {
urlsToCall()
.conflate() // to achieve the effect of switchMap
.transform { url ->
if (url.scheme == "local") {
val data = grabFromMemory(url.path)
emit(data)
} else {
withContext(Dispatchers.IO) {
performNetworkCall(url)
}
}
}
.collect {
// this will always execute in the main thread
// because this is where we collect,
// inside withContext(Dispatchers.Main)
}
}
}
协程代码将在它们被执行的上下文中运行。并且只有网络调用的部分会在 IO 线程上运行，而我们在这里看到的所有其他内容都将在主线程上运行。
好吧，实际上，我们不知道里面的代码grabFromMemory()会在哪里运行，如果它是一个挂起函数，我们只知道它会在主线程中被调用，但是在那个挂起函数中，我们可以使用另一个 Dispatcher，但是它什么时候会被调用返回结果，val data这将再次出现在主线程中。
这意味着，查看一段代码，更容易判断它将在哪个线程中运行，如果您看到显式的 Dispatcher = 就是那个调度程序，如果您没有看到它：在任何线程调度程序中，您正在查看的挂起调用正在被调用。
这不是 kotlin 发明的概念，但这是他们比我所知道的任何其他语言都更接受的东西。
那是什么？
使用 RxJava，您可以订阅 observable，它们会给您一个Disposable对象。
当不再需要它时，您需要处理它。所以你通常做的是保留对它的引用（或将它放在 a 中CompositeDisposable），以便以后dispose()在不再需要时调用它。如果你不这样做，linter 会给你一个警告。
RxJava 比传统线程好一些。当您创建一个新线程并在其上执行某些操作时，这是“一劳永逸”，您甚至没有办法取消它：Thread.stop()已被弃用、有害，而且最近的实现实际上什么也没做。Thread.interrupt()使您的线程失败等.. 任何异常都会丢失.. 你得到了图片。
使用 kotlin 协程和流程，它们颠倒了“一次性”概念。如果没有CoroutineContext.
这个上下文定义了scope你的协程。在其中产生的每个子协程都将共享相同的范围。
如果您订阅流，则必须在协程内或也提供范围。
您仍然可以保留对您启动的协程 ( Job) 的引用并取消它们。这将自动取消该协程的每个子进程。
如果您是 Android 开发人员，他们会自动为您提供这些范围。示例：viewModelScope并且您可以在具有该范围的视图模型中启动协程，知道它们将在视图模型被清除时自动取消。
viewModelScope.launch {
// my coroutine here
}
如果任何孩子失败，一些范围将终止，其他范围将让每个孩子离开自己的生命周期而不停止其他孩子如果失败（SupervisedJob）。
为什么这是一件好事？
让我试着像Roman Elizarov那样解释它。
一些旧的编程语言有这个概念，goto它基本上可以让您随意从一行代码跳转到另一行代码。
非常强大，但如果被滥用，你最终可能会得到非常难以理解的代码，难以调试和推理。
所以新的编程语言最终将它完全从语言中删除。
当您使用iforwhile或when在代码上进行推理时更容易：无论这些块内发生了什么，您最终都会从它们中出来，这是一个“上下文”，您没有奇怪的进出.
启动一个线程或订阅一个 RxJava observable 类似于 goto：您正在执行的代码将继续运行，直到“其他地方”停止。
对于协程，通过要求您提供上下文/范围，您知道当您的范围结束时，协程将在您的上下文完成时完成，无论您有单个协程还是 1 万个协程。
您仍然可以通过使用“转到”协程GlobalScope，出于同样的原因，您不应该goto在提供它的语言中使用协程。
Flow 仍在开发中，现在 RxJava 中可用的一些功能在 Kotlin Coroutines Flow 中仍然不可用。
大遗漏，现在，是share()运营商和它的朋友（publish()，replay()等...）
它们实际上处于高级开发状态，预计很快就会发布（在已经发布的 kotlin 之后不久1.4.0），您可以在这里看到 API 设计：
小智 6
您链接的演讲/文档不谈论频道。通道填补了您当前对协程的了解与事件驱动的编程之间的空白。
使用协程和通道，您可以像使用rx一样进行事件驱动的编程，但是您可以使用具有同步外观的代码进行操作，而无需使用许多“自定义”运算符。
如果您想更好地理解这一点，我建议您去看看Kotlin，因为这些概念更加成熟和完善（不是实验性的）。查看core.asyncClojure，Rich Hickey的视频，帖子和相关讨论。
| 归档时间： |
|
| 查看次数： |
16812 次 |
| 最近记录： |