kotlin - 生产者内部的监听器

Question

我正在尝试创建 producer对于听众。
我的代码看起来像这样

suspend fun foo() = produce{
    someEvent.addListener {
        this.send(it)
    }
}

但我收到错误 Suspension functions can be called only within coroutine这是有道理的。我的问题是。有没有办法使用协程来实现这种模式？

Answer 1

有几种方法可以实现它，具体取决于您要实现的目标:

如果您只想接收最近的事件，那么您应该使用合并 channel 和 offer方法为它成功:

fun foo() = produce<T>(capacity = Channel.CONFLATED) {
    someEvent.addListener {
        offer(it)
    }
}

如果接收所有事件至关重要，那么您的选择取决于事件生产者的行为。这里要思考的关键问题是，如果您的事件制作者开始“不间断”地产生大量事件，会发生什么。根据经验，大多数“同步”事件生产者不支持显式背压信号，但它们仍然支持隐式背压信号——如果它们的监听器速度较慢或阻塞线程，它们会减慢速度。因此，通常，以下解决方案非常适契约(Contract)步事件生产者:

fun foo() = produce<T>() {
    someEvent.addListener {
        runBlocking { send(it) }
    }
}

您还可以指定一些积极的 capacity = xxx作为 produce 的参数如果您遇到一次生成一批事件并且您不想阻止生产者但让消费者按照自己的节奏处理它们的情况，则构建器作为性能优化。

在极少数情况下，当您的生产者不理解隐式阻塞背压信号时(当它作为某种多线程装置在没有内部同步的情况下猛烈地产生事件时)，那么您可以使用具有无限容量的 channel offer ，但请注意，如果生产者的运行速度超过消费者，则可能会耗尽内存:

fun foo() = produce<T>(capacity = Channel.UNLIMITED) {
    someEvent.addListener {
        offer(it)
    }
}

如果您的生产者支持明确的背压信号(如功能性 react 流)，那么您应该使用特殊的适配器将其背压信号正确传输到协程/从协程传输。 kotlinx.coroutines库有许多开箱即用的集成模块以及用于此目的的各种 react 库。见 here .

注:你 不应该 标记您的 foo功能与 suspend修饰符。调用 foo无论如何都不会挂起调用者。它只是立即(同步)启动生产者协程。

要了解有关协程和不同类型 channel 的更多信息，我强烈建议学习 the guide on kotlinx.coroutines .