Scala中如何将多服务副本响应流合并为单个Iterator？

要实现按条目完成顺序即时返回的合并Iterator，核心思路是用线程安全的队列作为缓冲区，让各个并行流的元素一准备好就进入队列，合并后的Iterator从队列中取元素，同时跟踪所有流的处理状态以确定何时终止。

完整实现代码 #

import java.util.concurrent.LinkedBlockingQueue
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}

// 假设Request和ResultItem为业务定义的类型
case class Request(id: Int)
case class ResultItem(value: String)

// 模拟RPC调用返回响应流
def getResult(input: Request): Iterator[ResultItem] = {
  Iterator(
    s"Item from ${input.id}-1",
    s"Item from ${input.id}-2",
    s"Item from ${input.id}-3"
  ).map(ResultItem)
}

def getResults(inputs: Seq[Request])(implicit ec: ExecutionContext): Iterator[ResultItem] = {
  // 线程安全阻塞队列，存储待输出的元素或状态标记
  val queue = new LinkedBlockingQueue[Either[Throwable, ResultItem]]()
  // 跟踪未处理完成的流数量
  var remainingStreams = inputs.size

  // 并行处理每个请求的流
  inputs.foreach { request =>
    Future { getResult(request) }.onComplete {
      case Success(iterator) =>
        // 异步遍历流，元素就绪即放入队列
        Future {
          iterator.foreach(item => queue.put(Right(item)))
          // 当前流处理完成，计数器减一
          remainingStreams -= 1
          // 放入流结束标记
          queue.put(Left(new NoSuchElementException("Stream exhausted")))
        }
      case Failure(ex) =>
        // 处理RPC调用失败的情况，可根据业务调整（如忽略/返回错误条目）
        queue.put(Left(ex))
        remainingStreams -= 1
    }
  }

  // 实现合并后的Iterator
  new Iterator[ResultItem] {
    private var cachedNext: Option[ResultItem] = None

    override def hasNext: Boolean = {
      cachedNext match {
        case Some(_) => true
        case None =>
          // 循环读取队列，直到拿到有效元素或所有流结束
          while (true) {
            queue.take() match {
              case Right(item) =>
                cachedNext = Some(item)
                return true
              case Left(ex: NoSuchElementException) =>
                // 流结束标记，检查是否所有流都处理完
                if (remainingStreams == 0) return false
              case Left(ex) =>
                // 抛出RPC调用异常，或自定义错误处理逻辑
                throw ex
            }
          }
          false // 不可达分支
      }
    }

    override def next(): ResultItem = {
      if (hasNext) {
        val item = cachedNext.get
        cachedNext = None
        item
      } else {
        throw new NoSuchElementException("All streams exhausted")
      }
    }
  }
}

关键实现细节 #

1. 线程安全缓冲区：用LinkedBlockingQueue保证多线程环境下元素的正确读写，避免竞态条件。
2. 即时输出：每个流的元素会被逐个放入队列，不需要等整个流处理完成，实现了"哪个条目先就绪先返回"的需求。
3. 流状态跟踪：通过remainingStreams计数器结合队列中的结束标记，确保Iterator在所有流处理完成且缓冲区为空时才终止。
4. 异常处理：RPC调用失败时会将异常放入队列，Iterator读取到异常时直接抛出，你也可以修改逻辑（如返回错误类型的ResultItem或忽略失败流）。

使用说明 #

• 需要隐式传入ExecutionContext（如scala.concurrent.ExecutionContext.global）用于Future调度。
• 该Iterator是阻塞式的，调用next()时如果队列无元素会阻塞，直到有新元素或所有流结束。

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