kotlin - 当消费者的作业被取消时关闭协程 channel

Question

我有一个简单的生产者和一个使用协程 channel 的消费者。这是一个简化版本:

class Producer {

  suspend fun start(): ReceiveChannel<String> {

    val channel = Channel<String>(Channel.UNLIMITED)

    // Asynchronous channel.send(it) from an object callback

    channel.invokeOnClose {
      // Channel is closed...
    }

    return channel
  }

}

class Consumer : CoroutineScope {

  private val producer = Producer()

  private val job = Job()
  override val coroutineContext = job + Dispatchers.Default

  fun start() {
    launch {
      val channel = producer.start()

      for (currentValue in channel) {
        // use currentValue
      }
    }
  }

  fun stop() {
    job.cancel()
  }

}

生产者创建一个 channel ，然后用异步作业中的值填充它。 Consumer 对其进行迭代并使用这些值。

我的期望是，当我从消费者调用 job.cancel() 时， channel 迭代器将抛出异常并且 channel 将关闭。 invokeOnClose 回调永远不会被调用。

我可以在Consumer中维护对 channel 的引用并执行channel.close()。我想知道是否有更聪明的解决方案。也许还有另一种迭代 channel 值的方法？谢谢？

编辑

看起来像使用

launch {
    val channel = producer.start()

    channel.consumeEach { currentValue ->
    // use currentValue
    }
}

就可以了。然而，consumeEach() 被标记为已过时。

Answer 1

您希望 job.cancel() 会传播到您的生产者，但 Producer 实际上与任何东西都没有关系。将函数标记为挂起不会使其成为协程。

这是使用结构化并发解决此问题的一种方法:

class Producer: CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = Job() + Dispatchers.Default

    suspend fun start() = produce<String> {
        channel.send("A")

        channel.invokeOnClose {
            println("Closed")
        }
    }
}

现在您的 Producer 已经知道 CoroutineScope。

由于我们使用 product，因此您无需像之前那样初始化 channel 。