Java同步死锁？

Question

我对 Java 并发性还很陌生，并且在尝试使用锁和监视器编写一个玩具问题时遇到了困难。问题的要点是我有一个具有 get 和 put 方法的类，本质上是线程消费和生成的容器。我一生都无法获得正确的同步，最终要么陷入死锁，要么出现 IllegalMonitorStateException。

package concurrency

object ThreadsMain extends App {
  val syncVar = new SyncVar[Int]()

  val producer = new Thread {
    override def run(): Unit = {
      for (x <- 1 to 15) {
        syncVar.synchronized {
          if (!syncVar.isEmpty) {
            syncVar.wait()
          } else {
            syncVar.put(x)
            syncVar.notify()
          }
        }
      }
    }
  }

  producer.run()

  val consumer = new Thread {
    this.setDaemon(true)

    override def run(): Unit = {
      while (true) {
        syncVar.synchronized {
          if (syncVar.isEmpty) {
            syncVar.wait()
          } else {
            println(syncVar.get())
            syncVar.notify()
          }
        }
      }
    }
  }

  consumer.run()

  producer.join()

  consumer.join()
}

class SyncVar[T]() {
  var isEmpty: Boolean = true
  var value: Option[T] = None

  def get(): T = {
    if (isEmpty) throw new Exception("Get from empty SyncVar")
    else {
      val toReturn = value.get
      value = None
      isEmpty = true
      toReturn
    }
  }

  def put(x: T): Unit = {
    if (!isEmpty) throw new Exception("Put on non-empty SyncVar")
    else {
      value = Some(x)
      isEmpty = false
    }
  }
}

Answer 1

有几个问题:

1. 您应该使用start而不是run。
2. 如果您使用join，则无需将线程设置为守护线程。
3. 当您在生产者中执行 if ... else 时，您只能得到奇数。它应该只是 if 和 if 之后的其余部分(实际上 while 是更好的做法)。

我认为这样代码可以满足您的要求:

object ThreadsMain extends App {
  val syncVar = new SyncVar[Int]()
  val isDone = new AtomicBoolean(false)

  val producer = new Thread {
    override def run(): Unit = {
      for (x <- 1 to 15) {
        syncVar.synchronized {
          while (!syncVar.isEmpty) {
            syncVar.wait()
          }
          syncVar.put(x)
          syncVar.notify()
        }
      }
      isDone.set(true)
    }
  }

  producer.start()

  val consumer = new Thread {

    override def run(): Unit = {
      while (!isDone.get()) {
        syncVar.synchronized {
          while (syncVar.isEmpty) {
            syncVar.wait()
          }
          println(syncVar.get())
          syncVar.notify()
        }
      }
    }
  }

  consumer.start()

  producer.join()

  consumer.join()
}

class SyncVar[T]() {
  var isEmpty: Boolean = true
  var value: Option[T] = None

  def get(): T = {
    if (isEmpty) throw new Exception("Get from empty SyncVar")
    else {
      val toReturn = value.get
      value = None
      isEmpty = true
      toReturn
    }
  }

  def put(x: T): Unit = {
    if (!isEmpty) throw new Exception("Put on non-empty SyncVar")
    else {
      value = Some(x)
      isEmpty = false
    }
  }
}