Kotlin - 协程

并发

import kotlinx.coroutines.*

fun main() = runBlocking {
val deferred: Deferred<Int> = async {
loadData()
}
println("waiting...")
println(deferred.await())
}

suspend fun loadData(): Int {
println("loading...")
delay(1000L)
println("loaded!")
return 42
}

    runBlocking用作常规函数和挂起函数之间阻塞世界和非阻塞世界之间的桥梁。 它用作启动顶级主协程的适配器。

如果存在一个延迟对象列表，则可以调用awaitAll来等待所有对象的结果：

import kotlinx.coroutines.*

fun main() = runBlocking {
val deferreds: List<Deferred<Int>> = (1..3).map {
async {
delay(1000L * it)
println("Loading $it")
it
}
}
val sum = deferreds.awaitAll().sum()
println("$sum")
}

获取贡献者列表可以改为如下代码 :

suspend fun loadContributorsConcurrent(req: RequestData): List<User> = coroutineScope {
val service = createGitHubService(req.username, req.password)
val repos = service
.getOrgRepos(req.org)
.also { logRepos(req, it) }
.bodyList()val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
async {
service.getRepoContributors(req.org, repo.name)
.also { logUsers(repo, it) }
.bodyList()
}
}
deferreds.awaitAll().flatten().aggregate()

async(Dispatchers.Default) { ... }

</div>
<div>launch(Dispatchers.Main) {
updateResults()
}

launch(Dispatchers.Default) {
 val users = loadContributorsConcurrent(service, req)
 withContext(Dispatchers.Main) {
 updateResults(users, startTime)
 }
}

launch { /* this: CoroutineScope */
}

使用GlobalScope.async时，从全局范围定义的协程都是独立的。它们的寿命仅受整个应用程序寿命的限制。 可以存储从全局范围开始的协程的引用，等待其完成或显式取消它，但是它不会像结构化的那样自动发生。

使用外部scope的Context

由coroutineScope或由coroutine构建器创建的新作用域始终从外部作用域继承上下文。 在这种情况下，调用suspended loadContributorsConcurrent的scope是外部的scope：

launch(Dispatchers.Default) { //外部scope
 val users = loadContributorsConcurrent(service, req)
 // ...
}

所有嵌套的协程都是从继承的上下文自动开始的。 dispatcher就是这种情况的一部分。 这就是为什么async启动的所有协程都使用默认dispatcher的上下文启动的原因：

suspend fun loadContributorsConcurrent(req: RequestData): List<User> = coroutineScope {
 // 该scope从外部scope继承上下文
 // ... 
 async { // 嵌套协程从继承的上下文开始
 // ...
 }
 // ...
}

使用结构化并发，当创建顶级协程时，可以一次指定主要上下文元素（例如dispatcher）。 所有嵌套的协程都继承上下文并仅在需要时对其进行修改。

当为UI应用程序（例如Android）使用协程编写代码时，通常的做法是默认使用CoroutineDispatchers.Main作为顶层协程，然后在需要在不同线程上运行代码时显式放置其他dispatcher。

Channels

   众所周知，编写具有共享可变状态的代码非常困难且容易出错。 通过通信共享信息而不是使用公共可变状态共享信息可以简化此过程。 协程可以通过channels相互通信。

channels允许在不同协程之间传递数据。 一个协程可以向channel发送一些信息，而另一个协程可以从该channel接收此信息：

发送(产生)信息的协程通常被称为生产者，而接收(消费)信息的协程被称为消费者。 在需要时，许多协程可以将信息发送到同一channel，许多协程也可以从该channel接收信息：

当许多协程从同一channel接收信息时，每个元素仅由一个使用者处理一次；处理意味着从会从channel中删除此元素。

可以认为channel类似于元素集合(直接的模拟将是一个队列:将元素添加到一端并从另一端接收)。 但是，有一个重要的区别：与集合不同，即使在同步版本中，通道也可以暂停send和receive操作。 当channel为空或已满(channel的大小可能受到限制，可能已满)时，会发生这种情况。

Channel通过三个不同的接口表示:SendChannel、ReceiveChannel和继承前两个接口的Channel。 通常，创建一个channel并将其作为SendChannel实例提供给生产者，以便只有他们可以发送，并作为ReceiveChannel实例提供给消费者，以便只有他们可以从中接收。

注意，send和接收receive都声明为suspend方法

interface SendChannel<in E> {
suspend fun send(element: E)
fun close(): Boolean
}

interface ReceiveChannel<out E> {
suspend fun receive(): E
}

interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

生产者可以关闭通道以指示不再有元素。

Library中定义了几种类型的Channel。 它们在内部可以存储多种元素,只是在send调用是否可以挂起方面有所不同。 对于所有通道类型，receive调用的行为方式相同:如果通道不为空，则接收元素，否则将挂起。

1、Unlimited channel

无限制通道(Unlimited channel)是最接近队列的模拟:生产者可以将元素发送到此通道，并且它将无限增长。 send方法将永远不会被挂起。 如果没有更多的内存，则会抛出OutOfMemoryException。 和队列不同的是当使用者尝试从空通道接收消息并被挂起直到有一些新元素发送到该通道时继续使用。

2、Buffered channel

缓冲通道(Buffered channel)的大小受指定数字的限制。 生产者可以将元素发送到此通道，直到达到最大限制。 所有元素都在内部存储。 通道已满时，下一个send呼叫将被挂起，直到出现更多可用空间。

3、Rendezvous channel

"约定"通道(Rendezvous channel)是没有缓冲区的通道。 这与创建大小为零的缓冲通道(Buffered channel)相同。 其中一个功能(send或receive)始终被挂起，直到调用另一个功能为止。 如果调用了send函数，但消费者没有准备好处理该元素则receive会挂起，并且send也会被挂起。 同样，如果调用了receive函数且通道为空,换句话说,没有准备好发送该元素的的send被挂起-receive也会被挂起。

4、Conflated channel

发送到合并通道(Conflated channel)的新元素将覆盖先前发送的元素，因此接收方将始终仅能获取最新元素。 send调用将永远不会被挂起。

创建通道时，指定其类型或缓冲区大小(如果需要缓冲的通道)：

val rendezvousChannel = Channel<String>()
val bufferedChannel = Channel<String>(10)
val conflatedChannel = Channel<String>(CONFLATED)
val unlimitedChannel = Channel<String>(UNLIMITED)

   默认情况下，会创建一个"约定"通道(Rendezvous channel)。

  在以下示例中，将创建一个"约定"通道，两个生产者协程和一个消费者协程:

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*fun main() = runBlocking<Unit> {
val channel = Channel<String>()
launch {
channel.send("A1")
channel.send("A2")
log("A done")
}
launch {
channel.send("B1")
log("B done")
}
launch {
repeat(3) {
val x = channel.receive()
log(x)
}
}
}

fun log(message: Any?) {
println("[${Thread.currentThread().name}] $message")
}

以上将会打印如下结果:

[main @coroutine#4] A1
[main @coroutine#4] B1
[main @coroutine#2] A done
[main @coroutine#3] B done
[main @coroutine#4] A2