Scala中的后台任务

Posted 2023-04-01

【中文标题】Scala中的后台任务

【英文标题】：Background task in Scala

【发布时间】：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 包中定义一些帮助程序

【参考方案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)

【讨论】：

我认为这是 Akka 早期版本的代码；要对更新的版本做同样的事情，请参阅doc.akka.io/docs/akka/current/java/scheduler.html

【参考方案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 你确定funcToRunnable 的f 参数是惰性的吗？

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