scala - Scala 中有 FIFO 流吗？

Question

我正在 Scala 中寻找 FIFO 流，即提供以下功能的东西

• immutable.Stream (可以是有限的流并记住已读取的元素)
• mutable.Queue (允许向 FIFO 添加元素)

流应该是可关闭的，并且应该阻止对下一个元素的访问，直到添加该元素或关闭流。

实际上，我有点惊讶的是集合库没有(似乎)包含这样的数据结构，因为在我看来它是一个非常经典的数据结构。

我的问题:

• 1) 我是否忽略了什么？是否已经有一个类提供此功能？

• 2) 好的，如果它没有包含在集合库中，那么它可能只是现有集合类的简单组合。然而，我试图找到这个简单的代码，但对于这样一个简单的问题，我的实现看起来仍然相当复杂。对于这样的 FifoStream 是否有更简单的解决方案？

  class FifoStream[T] extends Closeable {
  
  val queue = new Queue[Option[T]]
  
  lazy val stream = nextStreamElem
  
  private def nextStreamElem: Stream[T] = next() match {
      case Some(elem) => Stream.cons(elem, nextStreamElem)
      case None       => Stream.empty
  }
  
  /** Returns next element in the queue (may wait for it to be inserted). */
  private def next() = {
      queue.synchronized {
          if (queue.isEmpty) queue.wait()
          queue.dequeue()
      }
  }
  
  /** Adds new elements to this stream. */
  def enqueue(elems: T*) {
      queue.synchronized {
          queue.enqueue(elems.map{Some(_)}: _*)
          queue.notify()
      }
  }
  
  /** Closes this stream. */
  def close() {
      queue.synchronized {
          queue.enqueue(None)
          queue.notify()
      }
  }
  }
  

Paradigmatic 的解决方案(稍作修改)

感谢您的建议。我稍微修改了范例的解决方案，以便 toStream 返回一个不可变的流(允许可重复读取)，从而满足我的需求。为了完整起见，这里是代码:

import collection.JavaConversions._
import java.util.concurrent.{LinkedBlockingQueue, BlockingQueue}

class FIFOStream[A]( private val queue: BlockingQueue[Option[A]] = new LinkedBlockingQueue[Option[A]]() ) {
  lazy val toStream: Stream[A] = queue2stream
  private def queue2stream: Stream[A] = queue take match {
    case Some(a) => Stream cons ( a, queue2stream )
    case None    => Stream empty
  }
  def close() = queue add None
  def enqueue( as: A* ) = queue addAll as.map( Some(_) )
}

Answer 1

在 Scala 中，流是“函数式迭代器”。人们期望它们是纯粹的(没有副作用)和不可变的。在您的情况下，每次迭代流时都会修改队列(因此​​它不是纯粹的)。这可能会造成很多误解，因为迭代同一流两次会产生两个不同的结果。

话虽这么说，您应该使用 Java BlockingQueues，而不是滚动您自己的实现。它们在安全性和性能方面被认为得到了很好的实现。这是我能想到的最干净的代码(使用你的方法):

import java.util.concurrent.BlockingQueue
import scala.collection.JavaConversions._

class FIFOStream[A]( private val queue: BlockingQueue[Option[A]] ) {
  def toStream: Stream[A] = queue take match {
    case Some(a) => Stream cons ( a, toStream )
    case None => Stream empty
  }
  def close() = queue add None
  def enqueue( as: A* ) = queue addAll as.map( Some(_) )
}

object FIFOStream {
  def apply[A]() = new LinkedBlockingQueue
}