Scala 中的多线程——只处理不变性

Question

我有Scala的代码

class MyClass {
  private val myData: Map[String, MyClass2] = new HashMap[String, MyClass2]()

  def someMethod = {
      synchronized(myData) {
        val id = getSomeId
        if (myData.containsKey(id)) myData = myData.remove(id)
        else Log.d("123", "Not found!")
      }
  }

  def getSomeId = //....
}

我想知道，是否可以在不使用 synchronized 并且不涉及其他一些库（例如 Akka 或任何其他库（类）甚至内置的情况下）的情况下保持此代码线程安全Java 还是 Scala？

理想情况下，我只想通过使用不变性的概念（Java 中的final，如果你愿意的话）来实现线程安全。

更新：

class MyClass(myData: Map[String, MyClass2] = new HashMap[String, MyClass2]()) {

  def someMethod = {
      synchronized(myData) {
        val id = getSomeId
        if (myData.containsKey(id)) new MyClass(myData.remove(id))
        else {
           Log.d("123", "Not found!")
           this
         }
      }
  }

  def getSomeId = //....
}

Answer 1

只有将MyClass 设为不可变（并让它也只使用不可变数据结构），才能解决不可变问题。原因很简单：如果MyClass是可变的，那么你必须通过并发线程同步修改。

这需要不同的设计 - 导致 MyClass 实例“更改”的每个操作都将返回（可能）修改后的实例。

import collection.immutable._

class MyClass2 {
  // ...
}

// We can make the default constructor private, if we want to manage the
// map ourselves instead of allowing users to pass arbitrary maps
// (depends on your use case):
class MyClass private (val myData: Map[String,MyClass2]) {
  // A public constructor:
  def this() = this(new HashMap[String,MyClass2]())

  def someMethod(id: String): MyClass = {
    if (myData.contains(id))
      new MyClass(myData - id) // create a new, updated instance
    else {
      println("Not found: " + id)
      this // no modification, we can return the current
            // unmodified instance
    }
  }

  // other methods for adding something to the map
  // ...
}

Answer 2

如果你使用 scala 2.10 中的 TrieMap，这是一个无锁并发映射实现，你可以避免同步：

import scala.collection.concurrent.TrieMap

class MyClass2

class MyClass {
    private val myData = TrieMap[String, MyClass2]()
    def someMethod = myData -= getSomeId
    def getSomeId = "id"
}

Answer 3

我会推荐使用库，因为自己获得并发是困难。例如，您可以使用像TrieMap 这样的并发映射。请参阅上面的答案。

但让我们假设您出于教育目的想要手动执行此操作。使上述线程安全的第一步是使用不可变集合。所以不是

private val myData: Map[String, MyClass2] = new HashMap[String, MyClass2]()

你会使用

private var myData = Map.empty[String, MyClass2] 

（尽管这里有一个 var，但它的可变状态比上面的版本少。在这种情况下，唯一可变的东西是单个引用，而在上面的示例中，整个集合都是可变的）

现在您必须处理 var。您必须确保在所有其他线程上“看到”一个线程上的 var 更新。所以你必须将该字段标记为@volatile。如果您有一个仅从一个线程完成写入的发布/订阅方案，那就足够了。但是假设您想从不同的线程读取和写入，您将需要对所有 write 访问使用同步。

显然，这足以保证引入一个小助手类：

final class SyncronizedRef[T](initial:T) {
  @volatile private var current = initial

  def get:T = current

  def update(f:T=>T) {
    this synchronized {
      current = f(current)
    }
  }
}

有了这个小助手，上面的代码可以这样实现：

class MyClass {
  val state = new SyncronizedRef(Map.empty[String, MyClass2])

  def someMethod = {
    state.update(myData =>
      val id = getSomeId
      if (myData.containsKey(id)) 
        myData - id
      else { 
        Log.d("123", "Not found!")
        myData
      }
  }

  def getSomeId = //....
}

就地图而言，这将是线程安全的。然而，整个事情是否是线程安全的取决于 getSomeID 中发生的任何事情。

一般来说，只要传递给 update 的东西是一个 pure 函数，它只是转换数据而没有任何副作用，这种处理并发的方法就可以工作。如果您的状态比单个地图更复杂，那么以纯函数式编写更新可能会非常具有挑战性。

SynchronizedRef 中仍然存在低级多线程原语，但您的程序逻辑完全没有它们。您只需通过组合纯函数来描述程序的状态如何响应外部输入而变化。

无论如何，这个特定示例的最佳解决方案就是使用现有的并发映射实现。

Answer 4

每当您共享可变状态时，您都需要一种并发机制。正如 Rüdiger 指出的那样，最好的办法是确定您拥有哪种并发场景，然后使用最适合该场景的现有工具：

• 普通的旧 Java 同步锁
• 原子比较和交换
• 软件事务内存（例如 Scala-STM）
• 消息传递/演员

当然，如果您不需要尽可能高的性能，您可以使用协作多任务（在单个共享线程上运行并发进程）。

如果您的类的状态是可监督的，您可以使该类完全不可变，例如：

case class MyClass2()

case class MyClass(myData: Map[String, MyClass2] = Map.empty) {
  def someMethod = {
    val id = getSomeId
    if (myData.contains(id)) copy(myData = myData - id)
    else throw new IllegalArgumentException(s"Key $id not found")
  }

  def getSomeId = "foo" // ...
}

MyClass 的实例仅通过将数据复制到新实例来进行变异，因此多个线程可以安全地引用同一个实例。但另一方面，如果两个线程 A 和 B 从同一个实例 foo1 开始，并且它们中的任何一个对其进行了变异并且 希望另一个线程看到该变异，那么你需要以某种形式再次共享该突变状态（使用 STM 参考单元，通过参与者发送消息，将其存储在同步变量中等）

val foo1 = MyClass(Map("foo" -> MyClass2()))
val foo2 = foo1.someMethod  // foo1 is untouched