用于多线程的 Scala 迭代器答案

【问题标题】：Scala Iterator for multithreading用于多线程的 Scala 迭代器
【发布时间】：2020-08-31 17:11:18
【问题描述】：

我在 synchronized 块中使用 scala Iterator 等待循环：

anObject.synchronized {
    if (Try(anObject.foo()).isFailure) {
        Iterator.continually { 
            anObject.wait()
            Try(anObject.foo()) 
        }.dropWhile(_.isFailure).next()
    }
    anObject.notifyAll()
}

是否可以将Iterator 与并发和多线程一起使用？如果不是，为什么？然后使用什么以及如何使用？

有一些细节，如果重要的话。 anObject 是一个可变队列。并且队列中有多个生产者和消费者。所以上面的块是这样的生产者或消费者的代码。 anObject.foo 是一个常见的简化函数声明，用于将数据入队（对于生产者）或出队（对于消费者）数据到队列或从队列中取出。

【问题讨论】：

标签： multithreading scala iterator thread-safety synchronized

【解决方案1】：

Iterator 在内部是可变的，所以如果你在多线程环境中使用它，你必须考虑到这一点。如果你保证你不会在以下情况下结束，例如

2个线程检查hasNext()
其中一个调用next() - 它恰好是最后一个元素
其他调用next() - NPE

（或类似的）那么你应该没问题。在您的示例中，Iterator 甚至没有离开范围，因此错误不应该来自 Iterator。

但是，在您的代码中，我看到 aObject.wait() 和 aObject.notifyAll() 彼此相邻的问题 - 如果您调用 .wait，那么您将无法到达 .notifyAll，这将取消阻止它。您可以在 REPL 中检查它是否挂起：

@ val anObject = new Object { def foo() = throw new Exception }
anObject: {def foo(): Nothing} = ammonite.$sess.cmd21$$anon$1@126ae0ca

@ anObject.synchronized {
      if (Try(anObject.foo()).isFailure) {
          Iterator.continually {
              anObject.wait()
              Try(anObject.foo())
          }.dropWhile(_.isFailure).next()
      }
      anObject.notifyAll()
  }
// wait indefinitelly

我建议将设计更改为不依赖wait 和notifyAll。但是，从您的代码中很难说出您想要实现的目标，所以我无法判断这是否更像 Promise-Future 案例、monix.Observable、monix.Task 或其他。

如果您的用例是一个队列、生产者和消费者，那么它听起来像是反应流的用例 - 例如。 FS2 + Monix，但它可能是 FS2+IO 或来自 Akka Streams 的东西

val queue: Queue[Task, Item] // depending on use case queue might need to be bounded

// in one part of the application
queue.enqueu1(item) // Task[Unit]

// in other part of the application
queue
  .dequeue
  .evalMap { item =>
    // ...
    result: Task[Result]
  }
  .compile
  .drain

这种方法需要对设计应用程序的想法进行一些改变，因为您将不再直接在线程上工作，而是设计一个流数据并声明什么是顺序的，什么可以并行完成，其中线程只是一个实现细节。

【讨论】：

谢谢 (+1)。我已经提供了一些详细信息。