Scala 和 Akka：后台工作答案

【问题标题】：Scala and Akka: Background jobScala 和 Akka：后台工作
【发布时间】：2017-01-11 20:24:49
【问题描述】：

我目前正在开发一个带有后台作业的 Play 应用程序，该应用程序应该定期发送我想使用 Akka 的邮件。我必须补充一点，我对 Scala/Play/Akka 真的很陌生。

目前我有以下设置：

// JobModule.scala
bind(classOf[MailJobScheduler]).asEagerSingleton()

这应该会启动下面的代码，它每秒都在运行

// MailJobScheduler.scala
val mailActor = actorSystem.actorOf(MailActor.props, "mail-actor")

actorSystem.scheduler.schedule(0 seconds, 1 seconds) {
    // check how many mails have to be sent and sent messages to the mailActor 
}

可能每秒都应该发送多封新邮件。我想知道：如果我每秒向 mailActor 发送 10 条消息，是否真的只有一个演员必须完成所有工作，还是会有多个演员同时工作？

如果是一个演员，我怎么能有多个演员可以分配工作？我可以/应该有多少？

【问题讨论】：

标签： scala playframework akka

【解决方案1】：

改用 Akka 流怎么样？

import akka.Done
import akka.stream.{KillSwitch, KillSwitches, OverflowStrategy}
import akka.stream.scaladsl.{Keep, Sink, Source}
import scala.concurrent.duration._
import scala.concurrent.Future

object BatchEmailSender {
  sealed trait Msg
  case object Tick extends Msg
  case class Email(toAddress: String, body: String) extends Msg

  def apply(sendEmail: Email => Future[Done], sendInterval: FiniteDuration = 10.seconds)(implicit mat: ActorMaterializer)
    : (Email => Unit, KillSwitch) = {
    val emailQueue = scala.collection.mutable.Queue[Email]()

    val (emailCmdQueue, killSwitch) = Source.queue[Msg](0, OverflowStrategy.backpressure)
      .merge(Source.tick(0.seconds, sendInterval, Tick))
      .viaMat(KillSwitches.single)(Keep.both)
      .toMat(Sink.foreach {
        case newEmail: Email =>
          emailQueue.enqueue(newEmail)
        case Tick =>
          emailQueue.dequeueAll(_ => true).foreach { email =>
            sendEmail(email).onFailure { case e =>
              println(s"Error sending email to ${email.toAddress}: $e")
            }
          }
      })(Keep.left)
      .run()

    (emailCmdQueue.offer(_), killSwitch)
  }
}

你需要一个 sendEmail 函数，然后它会像这样工作：

import scala.concurrent.ExecutionContext.Implicits.global // TODO: remove me

object TestApp extends App {
  import BatchEmailSender._
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  def sendEmail(email: Email): Future[Done] ={
    println(s"Sending email $email") // TODO: insert real email sender code here
    Future.successful(Done)
  }

  val (sendEmailEvery10s, killSwitch) = BatchEmailSender(sendEmail)
  sendEmailEvery10s(Email("foo@bar.com", "Email will arrive in 10s"))
  sendEmailEvery10s(Email("foo@bar.com", "Email will arrive in same batch"))
  Thread.sleep(11000)
  sendEmailEvery10s(Email("foo@bar.com", "Email will arrive after another 10s"))
  Thread.sleep(11000)
  killSwitch.shutdown()
}

我可能只是让你的生活复杂化了，但是 Akka 流让你可以做这些事情，而不必担心哪个演员做什么，有背压并且通常是更健壮的代码。

如果 Akka 流不存在，我会使用 1 个演员。累积actor中的所有消息，然后定期向自身发送一个滴答声。

【讨论】：

【解决方案2】：

按照您在示例中所做的那样使用调度程序，但我看不出mailActor 在这里如何帮助您。

actorSystem.scheduler.schedule(0 seconds, 1 seconds) {
    // just call the code the the checks for email
}

不要假设会有一个线程。即be extra careful to not close over unstable references

【讨论】：