如何在Scala中的线程之间进行通信？

Question

我在 Scala 中有一个包含一些方法的类，每个方法都有一个执行时间，例如 methodA 需要 10 秒，methodB 需要 5 秒。并且每个方法都异步调用。当我调用methodB 时，它应该取消正在运行另一个方法的线程。我首先调用methodA，2秒后调用methodB。这个问题的最佳解决方案是什么？

def methodA()={
    async{
        // a job that takes 10 seconds
    }
}
def methodB()={
   async{
       // other methods should stop their job
      // a job that takes 5 second
   }
}

def async[T](fn: => Unit): Unit = scala.actors.Actor.actor {
    fn
}

.....

methodA()
methodB()

Answer 1

这是一个想法，基于您的方法正在积极检查它是否仍应运行或取消的假设：

import concurrent.{ExecutionContext, Future, Promise, blocking, future, promise}

case class Cancelled() extends RuntimeException

object Job {
  def apply[A](fun: (() => Boolean) => A)(implicit ctx: ExecutionContext): Job[A] = 
    new Job[A] {
      private val p = promise[A]
      def result = p.future
      def cancel(): Unit = p.tryFailure(Cancelled())

      p tryCompleteWith future {
        fun(() => !p.isCompleted)
      }
    }
}
trait Job[A] {
  def result: Future[A]
  def cancel(): Unit
}

所以Job 与cancel() 方法一起体现了未来。您的示例可能与此类似：

import ExecutionContext.Implicits.global

val a = Job { active =>
  for (i <- 1 to 100 if active()) {
    blocking {
      Thread.sleep(1000)  // doing actual heavy work here
      println(s"A $i")
    }
  }
}

val b = Job { active =>
  for (i <- 1 to 20 if active()) {
    blocking {
      Thread.sleep(1000)  // doing actual heavy work here
      println(s"B $i")
    }
  }
  println("Goodbye A...")
  a.cancel()
}

我还看到一个相当 harsh variant 调用 Thread.interrupt。