如何避免使用Scala的Actor时的竞争条件

Question

我正在写一段代码，当缓冲区（列表）增长到一定的大小时，它将填充mongoDB集合。

import scala.actors.Actor 
import com.mongodb.casbah.Imports._ 
import scala.collection.mutable.ListBuffer 

class PopulateDB extends Actor { 
    val buffer = new ListBuffer[DBObject] 
    val mongoConn = MongoConnection() 
    val mongoCol = mongoConn("casbah_test")("logs") 

    def add(info: DBObject = null) { 
    if (info != null) buffer += info 

    if (buffer.size > 0 && (info == null || buffer.length >= 1000)) { 
     mongoCol.insert(buffer.toList) 
     buffer.clear 
     println("adding a batch") 
    } 
    } 

    def act() { 
    loop { 
     react { 
     case info: DBObject => add(info) 

     case msg if msg == "closeConnection" => 
      println("Close connection") 
      add() 
      mongoConn.close 
     } 
    } 
    } 
} 

然而，当我运行下面的代码，斯卡拉偶尔会扔在“mongoCol.insert（buffer.toList）”行“ConcurrentModificationException的”。我很确定它与“mongoCol.insert”有关。我想知道代码中是否存在任何根本性错误。或者我应该使用类似Akka的“atomic {...}”来避免这个问题。

下面是完整的堆栈跟踪：

PopulateDB@7e859a68: caught java.util.ConcurrentModificationException 
java.util.ConcurrentModificationException 
    at java.util.LinkedHashMap$LinkedHashIterator.nextEntry(LinkedHashMap.java:373) 
    at java.util.LinkedHashMap$EntryIterator.next(LinkedHashMap.java:392) 
    at java.util.LinkedHashMap$EntryIterator.next(LinkedHashMap.java:391) 
    at org.bson.BSONEncoder.putObject(BSONEncoder.java:113) 
    at org.bson.BSONEncoder.putObject(BSONEncoder.java:67) 
    at com.mongodb.DBApiLayer$MyCollection.insert(DBApiLayer.java:215) 
    at com.mongodb.DBApiLayer$MyCollection.insert(DBApiLayer.java:180) 
    at com.mongodb.DBCollection.insert(DBCollection.java:85) 
    at com.mongodb.casbah.MongoCollectionBase$class.insert(MongoCollection.scala:561) 
    at com.mongodb.casbah.MongoCollection.insert(MongoCollection.scala:864) 
    at PopulateDB.add(PopulateDB.scala:14) 
    at PopulateDB$$anonfun$act$1$$anonfun$apply$1.apply(PopulateDB.scala:26) 
    at PopulateDB$$anonfun$act$1$$anonfun$apply$1.apply(PopulateDB.scala:25) 
    at scala.actors.ReactorTask.run(ReactorTask.scala:34) 
    at scala.actors.Reactor$class.resumeReceiver(Reactor.scala:129) 
    at PopulateDB.scala$actors$ReplyReactor$$super$resumeReceiver(PopulateDB.scala:5) 
    at scala.actors.ReplyReactor$class.resumeReceiver(ReplyReactor.scala:69) 
    at PopulateDB.resumeReceiver(PopulateDB.scala:5) 
    at scala.actors.Actor$class.searchMailbox(Actor.scala:478) 
    at PopulateDB.searchMailbox(PopulateDB.scala:5) 
    at scala.actors.Reactor$$anonfun$startSearch$1$$anonfun$apply$mcV$sp$1.apply(Reactor.scala:114) 
    at scala.actors.Reactor$$anonfun$startSearch$1$$anonfun$apply$mcV$sp$1.apply(Reactor.scala:114) 
    at scala.actors.ReactorTask.run(ReactorTask.scala:36) 
    at scala.concurrent.forkjoin.ForkJoinPool$AdaptedRunnable.exec(ForkJoinPool.java:611) 
    at scala.concurrent.forkjoin.ForkJoinTask.quietlyExec(ForkJoinTask.java:422) 
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.mainLoop(ForkJoinWorkerThread.java:340) 
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:325) 

感谢， 德里克

Answer 1

DBObject是不是线程安全;你正在发送一个DBObject和你的actor消息。它可能会在稍后再次被修改，这将导致并发修改问题。

我会建议先尝试在DBObject上使用clone()，因为它会进入actor，并将其放入缓冲区。它只是一个浅拷贝，但至少应该足以导致LinkedHashMap上的并发修改问题，这些问题支持DBObject上的键（依靠LHM保持排序）。

我想尝试：

def act() { 
    loop { 
     react { 
     case info: DBObject => add(info.clone()) 

     case msg if msg == "closeConnection" => 
      println("Close connection") 
      add() 
      mongoConn.close 
     } 
    } 
    } 

如果不行，看看其他地方要修改的DBOBJECT它发送到演员后。

Answer 2

为什么下面class？

class PopulateDB extends Actor 

你保留多个PupulateDB演员？我期望object PopulateDB extends Actor，这样一个演员就可以集中这个任务。

除此之外，问题似乎是casbah或mongodb本身。