【问题标题】:Background task in ScalaScala中的后台任务
【发布时间】:2011-06-13 18:27:25
【问题描述】:

我有一个缓存,我想定期检查和修剪。在 Java 中,我会执行以下操作:

new Thread(new Runnable() {
  void run() {
    while (true) { 
      Thread.sleep(1000);
      // clear the cache's old entries
    }
  }
}).start();

当然,我在将线程安全类型用作缓存时会遇到一些问题,但撇开这些不谈,我的问题很简单。 Scala 运行重复后台任务的方式是什么——您不想在应用程序的主线程中运行?

我已经使用了一些演员,我想我在这种情况下的问题是我没有任何东西可以生成一条消息,表明是时候清除缓存了。或者更确切地说,我能想象生成这些消息的唯一方法是创建一个线程来执行它......

编辑:我需要人们对答案进行投票——他们对我来说都很好

【问题讨论】:

  • 线程生成很大程度上取决于您的系统。对于每个这样的情况,你可能会花费很多钱。看看 ScheduledThreadPoolExecutor,你可以从 Scala 中使用它,并在 scala.concurrent 包中定义一些帮助程序

标签: multithreading scala


【解决方案1】:

有很多方法可以做到这一点,但我会做一些简单的事情,如下所示。

import scala.concurrent.ops._

spawn {
  while (true) { 
    Thread.sleep(1000);
    // clear the cache's old entries
  }
}

希望这会有所帮助。

【讨论】:

  • .. 并确保您处理异常
  • 我会选择这个,因为在这种情况下我想要一个新线程。
  • scala.concurrent.ops 对象现已弃用(自版本 2.10.0 起)。
  • @Naetmul,应该改用什么?
  • @Kai 能否提供一些示例链接?
【解决方案2】:

您可以使用Akka Scheduler,它允许您向执行该工作的(akka)演员发送重复消息。从文档中,只需使用:

import akka.actor.Scheduler

//Sends messageToBeSent to receiverActor after initialDelayBeforeSending and then after each delayBetweenMessages
Scheduler.schedule(receiverActor, messageToBeSent, initialDelayBeforeSending, delayBetweenMessages, timeUnit)

【讨论】:

【解决方案3】:

Futures 是一种无需显式启动新线程的简单方法

import scala.actors.Futures._

// main thread code here

future {
   // second thread code here
}

// main thread code here

【讨论】:

  • 期货是为一次性任务而非重复任务而设计的。对于重复性任务而言,开销很大且不必要。
  • 他明确表示“反复出现”。您的样本中没有重复出现的内容。抱歉,-1
  • @RexKerr:但是关于基于spawn 的已接受答案的弃用信息给出了用未来替换spawn 的方向。那么 Scala 2.10 的正确答案是什么? @Jus12:要完成这项工作,可能需要import scala.concurrent.ExecutionContext.Implicits.global。 @akauppi:将来我会接受while (true) { ... } 以某种方式实现经常性......
  • @RexKerr:我原以为这两种方法在技术上是等效的,在这种情况下,人们可能会争辩说语法简化是一个很大的优势。我猜您出于技术原因(我不知道)偏爱老式方式...
  • @bluenote10 - Futures 通常在线程池上运行,这些线程池不能扩展到所有可用线程的大小,并且有额外的开销。在使用更有限的资源的更昂贵的解决方案和使用更少有限资源的更便宜的解决方案之间进行选择时,我倾向于选择限制更少的解决方案。
【解决方案4】:

spawn 很好,但请注意您的示例代码也适用于 Scala:

new Thread(new Runnable() {
  override def run() {
    while (true) { 
      Thread.sleep(1000);
      // clear the cache's old entries
    }
  }
}).start();

只需使用隐式转换来清理它:

implicit def funcToRunnable(f: => ()) = new Runnable() { override def run() { f() } }

new Thread{
  while(true) {
    Thread.sleep(1000);
    // blah
  }
}.start()

【讨论】:

  • 在我的测试(Scala 2.10.3)中,隐式版本是阻塞的,但第一个不是。
  • @davips 你确定funcToRunnablef 参数是惰性的吗?
  • Intellij 抱怨f: => (),所以我输入了f: => Unit。应该有什么区别吧?
【解决方案5】:

使用 Actors 而不绑定线程:

import actors.{TIMEOUT, Actor}
import actors.Actor._

private case class Ping( client: Actor, update: Int )
private case class Pulse()
case class Subscribe( actor: Actor )
case class Unsubscribe( actor: Actor )

class PulseActor extends Actor {
  def act = eventloop {
        case ping: Ping => { sleep(ping.update); ping.client ! Pulse }
  }
  def sleep(millis: Long) =
    receiveWithin(millis) {
      case TIMEOUT =>
  }
}

class ServiceActor extends Actor {

  var observers: Set[Actor] = Set.empty
  val pulseactor = new PulseActor
  val update = 2000

  def act = {
    pulseactor.start
    pulseactor ! new Ping( this, update )
    loop {
      react {
        case sub: Subscribe => observers += sub.actor
        case unsub: Unsubscribe => observers -= unsub.actor
        case Pulse => pulse
      }
    }
  }


  def pulse {  
    //cpuload = CPUprofile.getCPUload.getOrElse{ List(0.0) }  //real work
    observers foreach { observer => observer ! "CPUloadReport( cpuload )" }
    pulseactor ! Ping(this, update)
  }
}

object Exercise extends App {
  val deamon = new ServiceActor
  deamon.start
}

【讨论】:

    【解决方案6】:

    从 Scala 2.11.x 开始,恕我直言,Akka 演员似乎是最好的方法。首先创建一个任务调度库:

    import akka.actor.ActorSystem
    import scala.language.postfixOps
    import scala.concurrent.duration._
    val actorSystem = ActorSystem()
    val scheduler = actorSystem.scheduler
    implicit val executor = actorSystem.dispatcher
    
    // do once after period millis
    def doOnce(fn: => Unit, period:Long) = scheduler.scheduleOnce(period milliseconds)(fn) 
    // do regularly every period millis starting now
    def doRegularly(fn: => Unit, period:Long) = scheduler.schedule(0 seconds, period milliseconds)(fn)
    // do regularly every period millis if whileFn is true, starting now 
    def doWhile(fn: => Unit, whileFn: => Boolean, period:Long) {
     if (whileFn) {
        fn
        doOnce(doWhile(fn, whileFn, period), period)
     }
    }
    

    然后将其用作:

    doRegularly({
      println("hello world!")
    }, 1000) // repeat every 1000 millisecs
    
    doWhile({
      println("Sleeping!")
    }, whileDaylight, 1000) // repeat every 1000 millisecs whileDaylight is true
    

    【讨论】:

      猜你喜欢
      • 2013-07-18
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2013-08-07
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多