Scala：使用 ListBuffer 追加并行执行不会产生预期的结果

Question

我知道我在使用 mutable.ListBuffer 时做错了，但我不知道如何解决它（以及对问题的正确解释）。

我简化了下面的代码以重现该行为。

我基本上是在尝试并行运行函数，以便在处理我的第一个列表时将元素添加到列表中。我最终“失去”了元素。

import java.util.Properties

import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}


import scala.concurrent.{ExecutionContext}
import ExecutionContext.Implicits.global


object MyTestObject {

  var listBufferOfInts = new ListBuffer[Int]() // files that are processed

  def runFunction(): Int = {
    listBufferOfInts = new ListBuffer[Int]()
    val inputListOfInts = 1 to 1000
    val fut = Future.traverse(inputListOfInts) { i =>
      Future {
        appendElem(i)
      }
    }
    Await.ready(fut, Duration.Inf)
    listBufferOfInts.length
  }

  def appendElem(elem: Int): Unit = {
    listBufferOfInts ++= List(elem)
  }
}

MyTestObject.runFunction()
MyTestObject.runFunction()
MyTestObject.runFunction()

返回：

res0: Int = 937
res1: Int = 992
res2: Int = 997

显然我希望1000 一直被返回。如何修复我的代码以保持“架构”但让我的 ListBuffer“同步”？

Answer 1

我不知道你说你简化了它的确切问题是什么，但你仍然有一个明显的竞争条件，多个线程修改一个可变集合，这非常糟糕。正如其他答案指出的那样，您需要一些锁定，以便只有一个线程可以同时修改集合。如果您的计算量很大，以同步方式将结果附加到缓冲区应该不会显着影响性能，但如果有疑问，请始终进行测量。

但不需要同步，您可以做其他事情，而不需要 vars 和可变状态。让每个Future 返回您的部分结果，然后将它们合并到一个列表中，实际上Future.traverse 就是这样做的。

import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.concurrent.ExecutionContext.Implicits.global

def runFunction: Int = {
  val inputListOfInts = 1 to 1000

  val fut: Future[List[Int]] = Future.traverse(inputListOfInts.toList) { i =>
    Future { 
      // some heavy calculations on i
      i * 4
    }
  }

  val listOfInts = Await.result(fut, Duration.Inf)
  listOfInts.size
}

Future.traverse 已经为您提供了一个包含所有结果的不可变列表，无需将它们附加到可变缓冲区。 不用说，您总会收到1000 的回复。

@ List.fill(10000)(runFunction).exists(_ != 1000) 
res18: Boolean = false

Answer 2

我不确定上面是否显示了您正在尝试正确执行的操作。也许问题在于您实际上正在共享一个您在 runFunction 中重新初始化的 var ListBuffer。

当我把它拿出来时，我会收集我正确预期的所有事件：

import java.util.Properties

import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, Future }

import scala.concurrent.{ ExecutionContext }
import ExecutionContext.Implicits.global

object BrokenTestObject extends App {

  var listBufferOfInts = ( new ListBuffer[Int]() )

  def runFunction(): Int = {
    val inputListOfInts = 1 to 1000
    val fut = Future.traverse(inputListOfInts) { i =>
      Future {
        appendElem(i)
      }
    }
    Await.ready(fut, Duration.Inf)
    listBufferOfInts.length
  }

  def appendElem(elem: Int): Unit = {
    listBufferOfInts.append( elem )
  }

  BrokenTestObject.runFunction()
  BrokenTestObject.runFunction()
  BrokenTestObject.runFunction()

  println(s"collected ${listBufferOfInts.length} elements")
}

如果您确实有同步问题，您可以使用类似以下的方法：

import java.util.Properties

import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, Future }

import scala.concurrent.{ ExecutionContext }
import ExecutionContext.Implicits.global

class WrappedListBuffer(val lb: ListBuffer[Int]) {
  def append(i: Int) {
    this.synchronized {
      lb.append(i)
    }
  }
}

object MyTestObject extends App {

  var listBufferOfInts = new WrappedListBuffer( new ListBuffer[Int]() )

  def runFunction(): Int = {
    val inputListOfInts = 1 to 1000
    val fut = Future.traverse(inputListOfInts) { i =>
      Future {
        appendElem(i)
      }
    }
    Await.ready(fut, Duration.Inf)
    listBufferOfInts.lb.length
  }

  def appendElem(elem: Int): Unit = {
    listBufferOfInts.append( elem )
  }

  MyTestObject.runFunction()
  MyTestObject.runFunction()
  MyTestObject.runFunction()

  println(s"collected ${listBufferOfInts.lb.size} elements")
}

Answer 3

变化

listBufferOfInts ++= List(elem)

到

synchronized {
    listBufferOfInts ++= List(elem)
}

让它发挥作用。可能会成为性能问题？我仍然对解释感兴趣，也许还有更好的做事方式！