我必须使用 hasNext() 和 next() 方法实现一个 Iterator 接口(interface)(由 Java API 定义),它应该返回源自异步处理的 HTTP 响应(使用 Akka actor 处理)的结果元素。
必须满足以下要求:
- 不要阻塞并等待异步操作完成,因为生成大型结果集可能需要一段时间(迭代器应在结果元素可用时立即返回)
- Iterator.next() 应该阻塞直到下一个元素可用(或者如果没有更多元素出现则抛出异常)
- Iterator.hasNext() 应该返回 true,只要还有更多元素(即使下一个元素尚不可用)
- 结果总数事先未知。结果生产者将在完成时发送特定的“结束消息”。
- 尽量避免使用 InterruptedException,例如当迭代器正在等待一个空队列但不会生成更多元素时。
我还没有研究过 Java 8 流或 Akka 流。但由于我基本上必须遍历队列(有限流),所以我怀疑是否有任何合适的解决方案。
目前,我的 Scala 实现 stub 使用 java.util.concurrent.BlockingQueue 并且看起来像这样:
class ResultStreamIterator extends Iterator[Result] {
val resultQueue = new ArrayBlockingQueue[Option[Result]](100)
def hasNext(): Boolean = ??? // return true if not done yet
def next(): Result = ??? // take() next element if not done yet
case class Result(value: Any) // sent by result producing actor
case object Done // sent by result producing actor when finished
class ResultCollector extends Actor {
def receive = {
case Result(value) => resultQueue.put(Some(value))
case Done => resultQueue.put(None)
}
}
}
我使用 Option[Result] 来指示结果流的结尾为 None。我尝试过查看下一个元素并使用“完成”标志,但我希望有更简单的解决方案。
奖励问题:
- 单元测试如何涵盖同步/异步实现,尤其是测试延迟结果生成?
- 如何使迭代器成为线程安全的?
最佳答案
以下代码将满足要求。 Actor 的字段可以在 Actor 的接收器中安全地修改。 所以resultQueue不应该在Iterator的域,而应该在Actor的域。
// ResultCollector should be initialized.
// Initilize code is like...
// resultCollector ! Initialize(100)
class ResultStreamIterator(resultCollector: ActorRef) extends Iterator[Result] {
implicit val timeout: Timeout = ???
override def hasNext(): Boolean = Await.result(resultCollector ? HasNext, Duration.Inf) match {
case ResponseHasNext(hasNext) => hasNext
}
@scala.annotation.tailrec
final override def next(): Result = Await.result(resultCollector ? RequestResult, Duration.Inf) match {
case ResponseResult(result) => result
case Finished => throw new NoSuchElementException("There is not result.")
case WaitingResult => next()// should be wait for a moment.
}
}
case object RequestResult
case object HasNext
case class ResponseResult(result: Result)
case class ResponseHasNext(hasNext: Boolean)
case object Finished
case object WaitingResult
case class Initialize(expects: Int)
// This code may be more ellegant if using Actor FSM
// Acotr's State is (beforeInitialized)->(collecting)->(allCollected)
class ResultCollector extends Actor with Stash {
val results = scala.collection.mutable.Queue.empty[Result]
var expects = 0
var counts = 0
var isAllCollected = false
def beforeInitialized: Actor.Receive = {
case Initialize(n) =>
expects = n
if (expects != 0) context become collecting
else context become allCollected
unstashAll
case _ => stash()
}
def collecting: Actor.Receive = {
case RequestResult =>
if (results.isEmpty) sender ! WaitingResult
else sender ! ResponseResult(results.dequeue())
case HasNext => ResponseHasNext(true)
case result: Result =>
results += result
counts += 1
isAllCollected = counts >= expects
if (isAllCollected) context become allCollected
}
def allCollected: Actor.Receive = {
case RequestResult =>
if (results.isEmpty) sender ! Finished
else sender ! ResponseResult(results.dequeue())
case HasNext => ResponseHasNext(!results.isEmpty)
}
def receive = beforeInitialized
}
关于java - 使用 hasNext() 和 next() 遍历异步生成的元素流,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/30138672/