如何避免在 kotlin 频道生产者中使用 sendBlocking？答案

【问题标题】：How to avoid using sendBlocking in kotlin channel producer?如何避免在 kotlin 频道生产者中使用 sendBlocking？
【发布时间】：2020-06-05 05:04:52
【问题描述】：

我正在使用 kotlin 通道来迁移数据库：我有 1 个生产者和多个写入数据库的处理器。生产者只是将批量文档发送到通道：

fun CoroutineScope.produceDocumentBatches(mongoCollection: MongoCollection<Document>) = produce<List<Document>> {
    var batch = arrayListOf<Document>()
    for ((counter, document) in mongoCollection.find().withIndex()) {
        if ((counter + 1) % 100 == 0) {
            sendBlocking(batch)
            batch = arrayListOf()
        }
        batch.add(document)
    }
    if (batch.isNotEmpty()) sendBlocking(batch) }
}

这是我的处理器的样子：

private fun CoroutineScope.processDocumentsAsync(
        documentDbCollection: MongoCollection<Document>,
        channel: ReceiveChannel<List<Document>>,
        numberOfProcessedDocuments: AtomicInteger
) = launch(Dispatchers.IO) {
    // do processing
}

这就是我在脚本中使用它们的方式：

fun run() = runBlocking {
        val producer = produceDocumentBatches(mongoCollection)
        (1..64).map { processDocumentsAsync(documentDbCollection, producer, count) }
}

那么在性能方面使用sendBlocking 可以吗？如果我只使用send，我会在一个协程中创建许多挂起函数，因为写入数据库比读取慢得多，我得到java.lang.OutOfMemoryError: Java heap space。我是否正确理解生产者阻塞Main 线程但它对性能很好，因为所有消费者都在 IO 线程上执行？

【问题讨论】：

不是send 导致您的OutOfMemory 问题。这是另外一回事。您不应该在此代码中需要 sendBlocking。没有理由。

标签： kotlin kotlin-coroutines

【解决方案1】：

也许我的理解不正确，但我认为您并没有在协程中创建许多挂起函数，不确定这样说是否有意义。由于您在使用produce 函数时没有定义容量，因此默认值为零并且它正在使用RendezvousChannel。它将暂停直到另一个协程调用receive。我认为你不需要sendBlocking。

我的猜测是您使用了两个多个消费者并且在内存中有两个多个Document 实例，这可能是OutOfMemoryError 的原因。每个Document 的大小是多少？你的 jvm 堆配置是什么？

我的建议：

使用send 而不是sendBlocking
将消费者数量减少到更小的数量，看看OutOfMemory 是否仍然存在。
使用clear() 擦除ArrayList 而不是创建新实例。

如果按照这些建议运行正常，请尝试增加消费者数量并检查是否一切正常。

【讨论】：