十一、kotlin的协程(三)——热流channel

渣渣→_→

913人浏览 · 2022-07-12 22:07:01

渣渣→_→ · 2022-07-12 22:07:01 发布

theme: cyanosis

热流`channel` —— 管道

管道是什么？

本质上是一个BlockingQueue阻塞队列，只不过多了个可以挂起的函数罢了

为什么要用管道?

我们可以把 kotlin 的 channel 当作 BlockingQueue , 但是 channel 使用的是 挂起函数 的 send 代替阻塞队列的 put, 用挂起函数 receive 代替阻塞队列的 take这样 channel 的优势就有了, 不会阻塞当前线程
channel 是一个允许单向信息传递的数据结构, 从管道的写入端写入数据到管道的读取端读取数据, 这些都是串行的, 它的顺序是不变的

什么是热流？

前面我们知道冷流是 flow，需要末端操作（可以看成是开关）才会开启 emit 的函数发射元素过来

而热流呢？

则不需要什么末端操作（开关），只要 sender 就一定会将元素发送出去，至于 receiver 端是否读取，那就不清楚了

管道怎么用?

管道 hello world

fun main(): Unit = runBlocking {
	val channel = Channel<Int>()
	launch {
		for (x in 1..5) {
			channel.send(x * x)
		}
		channel.close()
	}
	for (item in channel) {
		println(item)
	}
	println("Done")
}

首先，send 发送变量过去就会被挂起, 直到另一方调用 recevie , send 函数才能够重新发送变量过去, 反之, receive 也是, 如果对面没有 send 任何变量, 则会挂起

就把管道当作queue, 把send当作add, 把receive当作get, 然后在管道中加入get阻塞功能就是channel了

这里的 send 为什么会阻塞呢？因为 channel 管道的大小所限，这会在后续说明

其次，热流管道需要close，来表示没有更多的元素发送了，此时程序才会停止，否则会继续阻塞

对比下 flow

在 flow 中要实现上面这种方式要咋办？

fun main(): Unit = runBlocking {
	val flow = flow {
		for (x in 1..5) {
			emit(x)
		}
	}
	
	flow.collect {
		println(it)
	}
	println("Done")
}

会不会感觉更加的简单，还不需要 close ，也可以不需要 launch

啊？什么？ flow 不需要 launch？

可以看 flow 的源码

public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)

不过在 collect 函数中则需要挂起 suspend

public suspend fun collect(collector: FlowCollector<T>)

jetbrain明显是想要开发者更多的使用 flow 而不是 channel（至少目前是如此）

管道怎么迭代和关闭?

管道迭代

fun main(): Unit = runBlocking {
   val channel = Channel<Int>()
   launch {
      for (x in 1..5) {
         channel.send(x * x)
      }
   }
   // 当管道没有数据的时候, 就会阻塞等待
   for (y in channel) {
      log(y)
   }
   log("Done")
}

管道的关闭及其关闭状态的判断

@ExperimentalCoroutinesApi
fun main(): Unit = runBlocking {
   val channel = Channel<Int>()
   launch {
      for (x in 1..5) {
         if (channel.isClosedForSend) {
            log("send: close send:${channel.isClosedForSend}, receive: ${channel.isClosedForReceive}")
            break
         }
         channel.send(x)
      }
   }
   repeat(5) {
      if (it >= 3) {
         channel.close()
      }
      if (channel.isClosedForReceive) {
         log("receive: close send:${channel.isClosedForSend}, receive: ${channel.isClosedForReceive}")
         return@repeat
      }
      log(channel.receive())
   }
   log("Done")
}

管道的生产者和消费者模型

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.produceSquares() = produce {
	for (x in 1..5) send(x * x)
}

fun main(): Unit = runBlocking {
    val channel = produceSquares()
	channel.consumeEach(::println)
	println("Done")
}

管道的结果可以给另一个管道

fun CoroutineScope.produceNumber() = produce<Int> {
   repeat(10) {
      send(it)
   }
}

fun CoroutineScope.square(number: ReceiveChannel<Int>) = produce<Int> {
   for (i in number) {
      send(i * i)
   }
}

fun main(): Unit = runBlocking {
   val number = produceNumber()
   val receiver = square(number)
   repeat(10) {
      log(receiver.receive())
   }
   log("Done")
   coroutineContext.cancelChildren()
}

注意：我们在调用 send 函数后，执行该 send 的协程会被挂起，线程去其他地方运行别的任务去了，所以看起来 while（true） send（xxxx） 好像它会无限发送元素似的，但其实如果管道的大小 buffer 只有一个的话，这里的 send 也只会被执行一次，向管道中发送一个元素，然后管道就满了， send 函数就会被挂起，等到管道中的元素被 receive 读取，send函数才会立即再发一个元素到管道中

管道与无穷质数序列

@ExperimentalCoroutinesApi
fun CoroutineScope.numbersProducer(start: Int) = produce {
   var n = start
   while (true) send(n++)
}

@ExperimentalCoroutinesApi
fun CoroutineScope.filterPrimes(numbers: ReceiveChannel<Int>, prime: Int) =
   produce {
      for (x in numbers) {
         if (x % prime != 0) {
            send(x) // 这里会找到所有满足条件的元素，然后 send 接着协程被挂起，不会无限的读取 numbers 里面的数字
         }
      }
   }

@ExperimentalCoroutinesApi
fun main(): Unit = runBlocking {
   var numbers = numbersProducer(2)
   while (true) {
      val prime = numbers.receive()
      log("$prime")
      numbers = filterPrimes(numbers, prime)
      delay(1000)
   }
}

扇出（读取）

多个协程从一个管道中读取数据

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.produceNumbers() = produce {
	var x = 1
	while (true) {
		send(x++)
		delay(100)
	}
}

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
	for (i in channel) {
		println("${Thread.currentThread().name}: Processor #$id received $i")
	}
}


fun main(): Unit = runBlocking {
    val produceNumbers = produceNumbers()
	repeat(5) {
		launchProcessor(it, produceNumbers)
	}
	delay(1000)
	produceNumbers.cancel()
}

控制台打印出来不同的协程在做接受

main @coroutine#3: Processor #0 received 1
main @coroutine#3: Processor #0 received 2
main @coroutine#4: Processor #1 received 3
main @coroutine#5: Processor #2 received 4
main @coroutine#6: Processor #3 received 5
main @coroutine#7: Processor #4 received 6
main @coroutine#3: Processor #0 received 7
main @coroutine#4: Processor #1 received 8
main @coroutine#5: Processor #2 received 9

广播channel

一个管道广播数据给多个协程

这种方式和前面扇出方式不同之处在于：

前面是一堆数据，分发给多个协程，每个协程得到的数据不重复

这里是一堆数据复制出多个资源给多个协程，每个协程得到的数据重复

@ObsoleteCoroutinesApi
fun main(): Unit = runBlocking {
    val channel = BroadcastChannel<Int>(Channel.BUFFERED)
    launch {
        List(5) {
            delay(200)
            channel.send(it)
        }
        channel.close()
    }
    List(5) {
        launch {
            val receiveChannel = channel.openSubscription()
            for (i in receiveChannel) {
                log("received: $i")
            }
        }
    }.joinAll()
}

看输出就知道了

main @coroutine#3: received: 0
main @coroutine#4: received: 0
main @coroutine#5: received: 0
main @coroutine#6: received: 0
main @coroutine#7: received: 0
main @coroutine#3: received: 1
main @coroutine#4: received: 1
main @coroutine#5: received: 1
main @coroutine#6: received: 1
main @coroutine#7: received: 1
main @coroutine#3: received: 2
main @coroutine#4: received: 2
main @coroutine#5: received: 2
main @coroutine#6: received: 2
main @coroutine#7: received: 2
main @coroutine#3: received: 3
main @coroutine#4: received: 3
main @coroutine#5: received: 3
main @coroutine#6: received: 3
main @coroutine#7: received: 3
main @coroutine#3: received: 4
main @coroutine#4: received: 4
main @coroutine#5: received: 4
main @coroutine#6: received: 4
main @coroutine#7: received: 4

扇入（发送）

多个协程可以发送到同一个通道

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
	while (true) {
		delay(time)
		channel.send(s)
	}
}

fun main(): Unit = runBlocking {
	val channel = Channel<String>()
	launch { sendString(channel, "foo", 200L) }
	launch { sendString(channel, "BAR!", 500L) }
	repeat(6) {
		println(channel.receive())
	}
	currentCoroutineContext().cancelChildren()
}

foo
foo
BAR!
foo
foo
BAR!

通道缓冲区

我们可以给管道设置一个缓冲区, 如果不设置缓冲区的话, 每次管道只能放一个元素

fun main(): Unit = runBlocking {
	val channel = Channel<Int>(4) // 启动带缓冲的通道
	val sender = launch { // 启动发送者协程
		repeat(10) {
			println("Sending $it") // 在每一个元素发送前打印它们
			channel.send(it) // 将在缓冲区被占满时挂起
		}
	}
	// 没有接收到东西……只是等待……
	delay(1000)
	sender.cancel() // 取消发送者协程
}

然后你会发现, 每次传入管道都是 3 个元素, 然后再读取, 对记得上面管道缓冲区限定传入的参数是 2 , 但后面却能往管道里传入 3 个元素, 这里需要注意

Sending 0
Sending 1
Sending 2
Sending 3
Sending 4

produce 也是可以设置capacity的

@ExperimentalCoroutinesApi
public fun <E> CoroutineScope.produce(
    context: CoroutineContext = EmptyCoroutineContext,
    capacity: Int = 0,
    @BuilderInference block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> =
    produce(context, capacity, BufferOverflow.SUSPEND, CoroutineStart.DEFAULT, onCompletion = null, block = block)

管道是公平的: 先进先出

发送端和接受端的通道秉承先入先出原则

data class Ball(var hits: Int)

fun main() = runBlocking {
    val table = Channel<Ball>() // 一个共享的 table（桌子）
    launch { player("ping", table) }
    launch { player("pong", table) }
    table.send(Ball(0)) // 乒乓球
    delay(1000) // 延迟 1 秒钟
    coroutineContext.cancelChildren() // 游戏结束，取消它们
}

suspend fun player(name: String, table: Channel<Ball>) {
    for (ball in table) { // 在循环中接收球
        ball.hits++
        println("$name $ball")
        delay(300) // 等待一段时间
        table.send(ball) // 将球发送回去
    }
}

ping Ball(hits=1)
pong Ball(hits=2)
ping Ball(hits=3)
pong Ball(hits=4)

select

select 在很多地方都听过, 比如 linux 的 select 机制的多路复用, select 里头如果我没记错是用一个线程监控多个文件句柄(我记得极限是1024个?), 类似于起一条线程不断的轮询每个数组中的某个值, 判断是否有事件需要处理, 不过后续 linux 他们觉得 select 文件句柄和轮询设计不太好, 于是使用上了 poll和epoll

在 golang 中也有 select ，借助管道触发走自己的分支

`kotlin`的`select`是什么?

在 kotlin 中 select 的select是这样的:

从多个onXXX事件中挑选一个可用的且是最快的那一个事件执行

on 开头的函数这里我看成是 JavaScript DOM 中的那种事件函数, 比如 onClick 就是鼠标单击事件

看下面代码:

@ExperimentalCoroutinesApi
fun CoroutineScope.fizz() = produce {
   while (true) {
      delay(300)
      send("Fizz")
   }
}

@ExperimentalCoroutinesApi
fun CoroutineScope.buzz() = produce {
   while (true) {
      delay(500)
      send("Buzz")
   }
}

suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
   select<Unit> {
      fizz.onReceive { s -> log("fizz -> $s") }
      buzz.onReceive { s -> log("buzz -> $s") }
   }
}

@ExperimentalCoroutinesApi
fun main(): Unit = runBlocking {
   val fizz = fizz()
   val buzz = buzz()
   selectFizzBuzz(fizz, buzz)
   // fizz.cancel()
   // buzz.cancel()
   coroutineContext.cancelChildren()
}

fizz -> 'Fizz'
buzz -> 'Buzz!'
fizz -> 'Fizz'
fizz -> 'Fizz'
buzz -> 'Buzz!'
fizz -> 'Fizz'
buzz -> 'Buzz!'

在 select 中的 on 系列事件函数将会被 select 注册到内部，监控起来

多路 `channel` 复用

fun main(): Unit = runBlocking {
	val channels = listOf(Channel<String>(), Channel<String>())
	launch {
		delay(100)
		channels[0].send("777")
	}
	launch {
		delay(200)
		channels[1].send("888")
	}
	val res = select<Int> {
		channels.forEach { channel ->
			channel.onReceive { s ->
				s[0] - '0'
			}
		}
	}
	println(res)
	currentCoroutineContext().cancelChildren()
}

最终速度快的先执行, 打印了 7

而这里：

channel.onReceive调用的不是public val onReceive: SelectClause1<E>这个接口的public fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (Q) -> R)这个函数

而是 invoke 函数

图片.png

public operator fun <Q> SelectClause1<Q>.invoke(block: suspend (Q) -> R)

如果你不知道 onReceive 函数是什么参数，则可以使用 invoke

`select` 轮询 `onSend` 事件

@OptIn(ExperimentalCoroutinesApi::class)
private fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
	for (num in 1..10) {
		delay(100)
		select<Unit> {
			onSend(num) { // 这里默认将会发送一个 num 数值到管道中，所以下面不需要再次发送
				// it.send(num * 10) // 这段代码如果执行，将会再次往管道中发送一个数字
			}
			side.onSend(num) { // 自动发送 num 到管道中
				// it.send(num * 100)
			}
		}
	}
}

fun main(): Unit = runBlocking {
	val side = Channel<Int>()
	launch {
		side.consumeEach { println("Side channel has $it") }
	}
	produceNumbers(side).consumeEach {
		println("consuming $it")
		delay(250)
	}
	println("Done consuming")
	currentCoroutineContext().cancelChildren()
}

consuming 1
Side channel has 2
Side channel has 3
consuming 4
Side channel has 5
Side channel has 6
consuming 7
Side channel has 8
Side channel has 9
consuming 10
Done consuming

上面这段代码主要写了两个 channel 读写方面的问题

其中主管道为：

produceNumbers(side).consumeEach {
    println("consuming $it")
    delay(250)
}

侧管道为：

launch {
	side.consumeEach { println("Side channel has $it") }
}

其中需要注意：主管道有延迟 250 ms

所以在打印的时候会发现，侧管道打印的比较多一些

如果将delay(100)这段代码删除

效果会更加的明显

@OptIn(ExperimentalCoroutinesApi::class)
private fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
   for (num in 1..10) {
      select<Unit> {
         onSend(num) {
            // it.send(num * 10) // 第二次发送
         }
         side.onSend(num) {
           	// it.send(num * 100) // 第二次发送
         }
      }
   }
}

consuming 1
Side channel has 2
Side channel has 3
Side channel has 4
Side channel has 5
Side channel has 6
Side channel has 7
Side channel has 8
Side channel has 9
Side channel has 10
Done consuming

select 轮询各种事件

select 轮询 `onAwait`

private fun CoroutineScope.asyncString(time: Int) = async {
   delay(time.toLong())
   "Waited for $time ms"
}

private fun CoroutineScope.asyncStringsList(): List<Deferred<String>> {
   val random = Random(3)
   return List(12) {
      asyncString(random.nextInt(1000))
   }
}

在 select 轮询中选一个结果

@Test
fun test01() = runBlocking<Unit> {
   val list = asyncStringsList()
   val result = select<String> {
      list.forEach { deferred ->
         deferred.onAwait.invoke {
            it
         }
      }
   }
   log(result)
}

select每次只获得一个结果，这点需要注意

如果需要 select 获得更多的结果则代码可以改成如下：

fun main(): Unit = runBlocking {
   val list = asyncStringsList()
   list.withIndex().forEach { (index, deferred) ->
      val res = select<String> {
         deferred.onAwait.invoke {
            "Deferred $index produced answer '$it'"
         }
      }
      println(res)
   }
   val countActive = list.count { it.isActive }
   println("$countActive coroutines are still active")
}

总结下 `select`

select 类似于一个监控器, 不断监控其内部的所有事件, 直到有一个事件被触发, 直接执行, select 就不会再触发其他事件了, 除非再起一个select用于监控

    fun CoroutineScope.asyncString(str: String, time: Long) = async {
      delay(time)
      str
   }
   
   /**
    * 传入一个管道, 从管道中读取数据
    */
   @ExperimentalCoroutinesApi
   private fun CoroutineScope.switchDeferredChannel(input: Channel<Deferred<String>>) = produce {
      
      // 获取管道的第一个数据
      var current = input.receive()
      // 判断管道是否被关闭
      while (isActive) {
         // select 轮询内部事件
         select<Deferred<String>?> {
            // 设置 receiveCatching 事件, 只要管道有该事件就读取出来
            input.onReceiveCatching.invoke {
               it.getOrNull()
            }
            // 间前面事件读取出来的结果再注册一个轮询的 await 事件
            current.onAwait.invoke {
               // 说明我们的 asyncString 函数的 async 已经执行结果, 返回的 deferred 延迟值
               // awit 到我们 asyncString 函数直接的结果了
               send(it)
               // 再从管道中读取下一个 Deferred
               input.receiveCatching().getOrNull()
            }
         }?.let {
            // 把 select 读取出来的结果重新设置为 current 下次循环 再次在 select 中注册 onAwait 事件
            current = it
         } ?: break
      }
   }
   
   @Test
   fun test01() = runBlocking<Unit> {
      val channel = Channel<Deferred<String>>()
      launch {
         for (s in switchDeferredChannel(channel)) {
            log(s)
         }
      }
      channel.send(asyncString("BEGIN", 100))
      // 这里需要等待时间, 否则间之间被 cancel 掉
      delay(1000)
//    channel.close()
      coroutineContext.cancelChildren()
   }

协程异常处理

协程如果被取消就会在挂起点处抛出异常 CancellationException, 正常情况下我们是无法感受到的, 需要专门去捕捉它

挂起点: 是 suspend 函数 + 异步操作 的位置被叫做挂起点

suspend fun main() {
	coroutineScope {
		val job = launch {
			try {
				repeat(100000) {
					delay(1000)
					println("${Thread.currentThread()}: ping")
				}
			}
			catch (e: Exception) {
				e.printStackTrace()
			}
		}
		delay(3000)
		job.cancel()
	}
}

Thread[DefaultDispatcher-worker-1,5,main]: ping
Thread[DefaultDispatcher-worker-1,5,main]: ping
kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job=StandaloneCoroutine{Cancelling}@5d5390a2

上面的代码挂起点在 delay 这里，所以异常也将会在 delay 这里被发现和抛出

异常的传播

自动传播异常与向用户暴露异常

协程构造器有两种: 自动传播异常 (launch 和 actor)或向用户暴露异常 (async和produce), 如果协程构造器创建于一个根协程(即没有任何协程的子协程是它), 此时自动传播异常方式的异常将被视为未捕获异常, 异常将直接抛出 , 而向用户暴露异常依赖用户来最终消费异常(await或者receive)

自动传播异常的根协程会直接抛出异常

向用户暴露异常根据调用例如： await 或者 receive 这两个函数调用来发出异常

具体看代码

@OptIn(DelicateCoroutinesApi::class)
fun main(): Unit = runBlocking {
	val job = GlobalScope.launch {
		// 这个异常不需要捕获，直接就把异常抛出了
		throw IndexOutOfBoundsException()
	}
	job.join()
	val deferred = GlobalScope.async {
		// 这里虽然抛出异常，但是不会被执行
		throw ArithmeticException()
	}
//	try {
	// 上面的 throw ArithmeticException 在下面这行代码抛出异常
	// 如果我们没有捕获这个异常的话，将会抛出 ArithmeticException 异常
	deferred.await()
//	} catch (e: ArithmeticException) {
//		e.printStackTrace()
//	}
}

看抛出异常的行号

第一个异常在： throw IndexOutOfBoundsException() 这里抛出

第二个异常在：deferred.await()这里抛出异常，而不是throw ArithmeticException()这一行代码

`CoroutineExceptionHandler`

CoroutineExceptionHandler 仅对直接抛出的异常的协程进行捕获(launch这种不是async)

而它也仅对 CoroutineScope的上下文或者根协程中才能捕获

async 无法使用 CroutineExceptionHandler 的方式捕获异常，它只能够在 await 中发现异常

private val handler =  CoroutineExceptionHandler { _, throwable ->
	log("1 handler $throwable")
}

@DelicateCoroutinesApi
fun main(): Unit = runBlocking {
	val job = GlobalScope.launch(handler) {
		throw AssertionError() // 只会拦截此异常
	}
	val deferred = GlobalScope.async(handler) {
		throw ArithmeticException() // async 内的这一行代码 不会捕获异常
	}
	joinAll(job, deferred)
}

fun main() {
	val scope = CoroutineScope(handler)
	scope.launch {
		val job = launch {
			throw AssertionError() // 只会拦截此异常
		}
		val deferred = async {
			throw ArithmeticException() // async 内的这一行代码 不会捕获异常
		}
		joinAll(job, deferred)
	}
	TimeUnit.SECONDS.sleep(2)
}

但是这样不太方便, 因为需要在每个协程中添加上异常 handler

使用`META-INF`的配置文件实现全局的异常查看

他不会捕获异常, 仅仅是查看有什么异常, 合适做程序崩溃日志

val scope = CoroutineScope(handler)

GlobalScope.launch(handler)

所以我们还可以这样:

class GlobalCoroutinesException : CoroutineExceptionHandler {
	override val key: CoroutineContext.Key<*>
		get() = CoroutineExceptionHandler
	
	override fun handleException(context: CoroutineContext, exception: Throwable) {
		log("Coroutine exception: $exception")
	}
}

@DelicateCoroutinesApi
fun main(): Unit = runBlocking {
	val job = GlobalScope.launch {
		throw AssertionError()
	}
	val deferred = GlobalScope.async {
		throw ArithmeticException()
	}
	joinAll(job, deferred)
}

然后在 resouces 目录下创建 services 目录, 再在services目录下创建文件 kotlinx.coroutines.CoroutineExceptionHandler

在目录下填入 GlobalCoroutiesException 的包名和类型 coroutines18.exception.GlobalCoroutiesException

就可以运行 main 函数了

不过还是有问题

取消与异常

协程如果取消, 就会抛出一个 CancellationException , 但是我们无法发现该异常, 需要 catch 去捕获它, 才能发现

而如果子协程的 job 调用了 cancel 它不会取消掉父协程

fun main(): Unit = runBlocking {
    val child = launch {
        try {
            delay(Long.MAX_VALUE)
        } catch (e: Exception) {
            log("我捕获了异常$e")
        }
        finally {
            log("子协程被取消")
        }
    }
    yield()
    log("准备取消子协程: ")
    child.cancelAndJoin()
    yield()
    log("父协程结束")
}

如果协程遇到了 CancellationException以外的异常，它将使用该异常取消它的父协程。

这个行为无法被覆盖，并且用于为结构化的并发提供稳定的协程层级结构。

CoroutineExceptionHandler的实现并不是用于子协程。

在这些示例中，CoroutineExceptionHandler 总是被设置在由 GlobalScope 启动的协程中。将异常处理者设置在 runBlocking 主作用域内启动的协程中是没有意义的，尽管子协程已经设置了异常处理者，但是主协程也总是会被取消的。

异常聚合: 探讨多个子协程都抛出异常时会怎样?

private val handler = CoroutineExceptionHandler { _, throwable ->
    log("log: $throwable ${throwable.suppressed.get(0)}")
}

@DelicateCoroutinesApi
fun main(): Unit = runBlocking {
    val job = GlobalScope.launch(handler) {
        launch {
            try {
                    delay(Long.MAX_VALUE)
            }
            finally {
                    throw ArithmeticException()
            }
        }
        launch {
            try {
                delay(1000)
            }
            finally {
                throw ConcurrentModificationException()
            }
        }
    }
    job.join()
}

如代码显示, kotlin将会在异常的 suppressed 属性里放置多个异常

非根协程的异常总是会被逐层传播给根协程

异常的传播还涉及作用域间的问题, 如果我们前面使用的 GlobalScope 创建协程, 意味着该异常在独立的协程作用域中, 而我们使用的 coroutineScope 创建的协程不会这样

coroutineScope 创建的协程中如果子协程遇到异常, 他会往外抛出异常给其父协程, 父协程如果还有父协程, 则会继续传递给父协程的父协程, 直到传递给根协程处理

所以如果任何一个子协程出现了异常(除了CancellationException), 则会将其他协程都停止掉, 其他协程没有异常

这里出现了子协程出错整个协程树都结束的问题, 后面会出现解决的方法 supervisorScope

而 GlobalScope 创建的协程是独立于其所在的协程的父协程的, 可以说也是根协程, 所以不会出现这种问题

相当于 fork 了一个新的进程，但这里是创建了一个新的协程上下文，异常在协程上下文中传播的，最后都会传播到根协程，不在同一个协程上下文中，异常不可能传播到另一个上下文中

`SupervisorJob`: 防止子协程异常传播导致整棵协程树失败

fun main(): Unit = runBlocking {
	val scope = CoroutineScope(SupervisorJob())
	val job1 = scope.launch {
		delay(1000)
		throw RuntimeException()
	}
	val job2 = scope.launch {
		repeat(10) {
			delay(150)
			log("协程2 一直再运行")
		}
	}
	joinAll(job1, job2)
}

这样 job1 出现了异常, job2 也可以执行, 直到执行完毕

如果我们需要取消掉所有的协程可以调用 scope.cancel()

`supervisorScope`

我们也可以像下面这样用

suspend fun main(): Unit = supervisorScope {
	launch {
		repeat(10) {
			delay(200)
			log("子协程活着")
		}
	}
	launch {
		delay(1000)
		throw AssertionError()
	}
}

也能达到相同的效果

但是这种方式存在一个问题

如果作用域内存在作用域的异常, 比如下面这样:

suspend fun main(): Unit = supervisorScope {
	launch {
		repeat(10) {
			delay(200)
			log("子协程活着")
		}
	}
	delay(1000)
	throw AssertionError()
}

协程再也没机会执行 10 次循环了, 它会被 AssertionError 退出整个 supervisorScope 作用域空间

则会继续传递给父协程的父协程, 直到传递给根协程处理**

所以如果任何一个子协程出现了异常(除了CancellationException), 则会将其他协程都停止掉, 其他协程没有异常

这里出现了子协程出错整个协程树都结束的问题, 后面会出现解决的方法 supervisorScope

而 GlobalScope 创建的协程是独立于其所在的协程的父协程的, 可以说也是根协程, 所以不会出现这种问题

相当于 fork 了一个新的进程，但这里是创建了一个新的协程上下文，异常在协程上下文中传播的，最后都会传播到根协程，不在同一个协程上下文中，异常不可能传播到另一个上下文中

`SupervisorJob`: 防止子协程异常传播导致整棵协程树失败

fun main(): Unit = runBlocking {
	val scope = CoroutineScope(SupervisorJob())
	val job1 = scope.launch {
		delay(1000)
		throw RuntimeException()
	}
	val job2 = scope.launch {
		repeat(10) {
			delay(150)
			log("协程2 一直再运行")
		}
	}
	joinAll(job1, job2)
}

这样 job1 出现了异常, job2 也可以执行, 直到执行完毕

如果我们需要取消掉所有的协程可以调用 scope.cancel()

`supervisorScope`

我们也可以像下面这样用

suspend fun main(): Unit = supervisorScope {
	launch {
		repeat(10) {
			delay(200)
			log("子协程活着")
		}
	}
	launch {
		delay(1000)
		throw AssertionError()
	}
}

也能达到相同的效果

但是这种方式存在一个问题

如果作用域内存在作用域的异常, 比如下面这样:

suspend fun main(): Unit = supervisorScope {
	launch {
		repeat(10) {
			delay(200)
			log("子协程活着")
		}
	}
	delay(1000)
	throw AssertionError()
}

协程再也没机会执行 10 次循环了, 它会被 AssertionError 退出整个 supervisorScope 作用域空间

天工造物局

北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台

更多推荐

deck.gl TripsLayer 出租车轨迹动画示例实战：从示例运行到着色器原理

TripsLayer 是 deck.gl 中用于渲染**随时间流动的轨迹动画**（如出租车行程、物流路径、迁徙路线）的核心图层。本文以仓库中的最小独立示例 [examples/website/trips](https://link.gitcode.com/i/1442af8f442f4cff39ff546e050fceb5) 为骨架，带你完成从安装依赖、启动应用、理解曼哈顿出租车数据格式，到深入

天工造物局

Foundry Anvil 中 eth_simulateV1 的 Blob 交易与 Blob Gas 字段规范化：原理、实现与测试验证

本篇技术指南以 [anvil-simulate-blob-transactions.md](https://link.gitcode.com/i/3d2ebe227262de169335eb290af80da8) 这条 changelog 条目为起点，深入剖析 Foundry Anvil 对 `eth_simulateV1` RPC 返回结果中 blob（EIP-4844）交易与 blob gas

天工造物局

LMCache CLI 实战：用 `lmcache kvcache clear` 管理运行中服务的 KV Cache 状态

本文是一份针对 LMCache 官方 CLI 命令 `lmcache kvcache` 的完整技术指南。该命令用于在**运行中的 LMCache 服务（MP HTTP server）**上管理 KV cache 状态，当前提供 `clear` 子命令，用于清空 L1（CPU 内存）层中缓存的全部 KV 数据。读完本文，你将掌握该命令的完整参数体系、退出码语义、终端/JSON/静默三种输出模式，并能

天工造物局

所有评论(0)

查看更多评论

渣渣→_→

@qq_30690165

已为社区贡献2条内容

十一、kotlin的协程(三)——热流channel

渣渣→_→

theme: cyanosis

热流channel —— 管道

管道是什么？

为什么要用管道?

什么是热流？

管道怎么用?

管道 hello world

管道怎么迭代和关闭?

管道迭代

管道的关闭及其关闭状态的判断

管道的生产者和消费者模型

管道的结果可以给另一个管道

管道与无穷质数序列

扇出（读取）

广播channel

扇入（发送）

通道缓冲区

管道是公平的: 先进先出

select

kotlin的select是什么?

多路 channel 复用

select 轮询 onSend 事件

select 轮询各种事件

select 轮询 onAwait

总结下 select

协程异常处理

异常的传播

自动传播异常与向用户暴露异常

CoroutineExceptionHandler

使用META-INF的配置文件实现全局的异常查看

取消与异常

异常聚合: 探讨多个子协程都抛出异常时会怎样?

非根协程的异常总是会被逐层传播给根协程

SupervisorJob: 防止子协程异常传播导致整棵协程树失败

supervisorScope

SupervisorJob: 防止子协程异常传播导致整棵协程树失败

supervisorScope

所有评论(0)

温馨提示：您尚未绑定手机号

渣渣→_→

热流`channel` —— 管道

`kotlin`的`select`是什么?

多路 `channel` 复用

`select` 轮询 `onSend` 事件

select 轮询 `onAwait`

总结下 `select`

`CoroutineExceptionHandler`

使用`META-INF`的配置文件实现全局的异常查看

`SupervisorJob`: 防止子协程异常传播导致整棵协程树失败

`supervisorScope`

`SupervisorJob`: 防止子协程异常传播导致整棵协程树失败

`supervisorScope`