如何用LinkedBlockingQueue实现生产者消费者？避免消费任务误终止

问题背景 #

我遇到一个生产者消费者场景，代码实现如下：

生产者协程 #

suspend fun produceElements() {
  // 耗时操作获取元素
  val item = fetchItem()

  // 添加到队列
  linkedBlockingQ.add(item) 
}

消费者协程 #

suspend fun receiveItems() {
   while(true) {
      val item = linkedBlockingQ.take() // 等待第一个元素到来

      // 执行耗时任务处理元素

      if(linkedBlockingQ.isEmpty()) break
   }
}

主调用方法 #

suspend fun doMain() {
   val receiveTask = async { receiveItems() }

   val produceTask = async { produceElements() }
   produceTask.await() // 等待生产者完成
   receiveTask.await() // 等待消费者完成
}

我的担忧：当生产元素存在延迟时，LinkedBlockingQueue可能为空，导致消费任务错误地提前完成。请问当前实现方式是否正确？

结论：当前实现不正确 #

核心问题有两点：

1. 退出条件不可靠：消费者通过linkedBlockingQ.isEmpty()判断是否退出，完全无法区分"生产者暂时没产出"和"生产已完成"两种情况。只要队列短暂为空（比如fetchItem()耗时较长），消费者就会直接终止，错过后续生产的元素。
2. 缺少生产完成信号：生产者和消费者之间没有明确的结束约定，消费者永远无法确定是否还有后续元素需要处理。

修正方案 #

推荐方案：使用Kotlin协程原生Channel替代LinkedBlockingQueue #

Channel是协程专为跨协程通信设计的组件，自带关闭通知和安全的遍历逻辑，完全适配协程场景：

// 用Channel替代LinkedBlockingQueue，可指定容量（默认无界）
private val channel = Channel<Item>()

suspend fun produceElements() {
    // 模拟持续生产逻辑，根据实际需求调整循环条件
    while (hasMoreItemsToProduce()) {
        val item = fetchItem()
        channel.send(item) // 协程安全的发送操作
    }
    // 生产完成后关闭通道，通知消费者停止
    channel.close()
}

suspend fun receiveItems() {
    // 遍历通道，通道关闭后自动退出循环
    for (item in channel) {
        // 执行耗时的元素处理任务
        processItem(item)
    }
}

suspend fun doMain() = coroutineScope {
    // 启动消费者协程
    launch { receiveItems() }
    // 启动生产者协程并等待其完成
    launch { produceElements() }.join()
}

这个方案无需手动处理结束标记，协程会自动处理通道关闭后的遍历退出，彻底避免了原实现中的提前退出问题。

备选方案：给LinkedBlockingQueue添加结束标记（Poison Pill） #

如果必须使用LinkedBlockingQueue，可以通过添加特殊的"毒丸"标记来通知消费者生产完成：

// 定义结束标记，根据元素类型选择合适的标记（比如null或自定义对象）
private val POISON_PILL = null as Item?

suspend fun produceElements() {
    while (hasMoreItemsToProduce()) {
        val item = fetchItem()
        linkedBlockingQ.add(item)
    }
    // 生产完成后放入结束标记
    linkedBlockingQ.add(POISON_PILL)
}

suspend fun receiveItems() {
    while (true) {
        val item = linkedBlockingQ.take()
        // 收到结束标记则退出
        if (item == POISON_PILL) break
        // 处理元素
        processItem(item)
    }
}

注意：此方案需要确保所有生产者都发送结束标记，且消费者正确识别标记，否则可能出现死锁或遗漏元素的情况。

内容的提问来源于stack exchange，提问作者Renjith