Akka Typed如何给masterRegistryActor添加重复定时任务及实现全局引用

问题1 周期性定时调度找不到API的原因与解决方案 #

原因 #

你在Behaviors.setup内部无法直接调用外部的system.scheduler，是因为Behaviors.setup的逻辑会在ActorSystem初始化过程中执行，此时你外部声明的implicit val system还没有完成赋值，自然无法直接引用。

解决方案 #

Behaviors.setup传入的context参数自带系统引用，直接通过context.system.scheduler即可获取调度器，使用scheduleWithFixedDelay（固定间隔调度，任务执行时间不会叠加到间隔上）实现周期性发消息即可，示例代码如下：

import scala.concurrent.duration._

val rootBehavior = Behaviors.setup[Nothing] { context =>
  val masterRegistryActor = context.spawn(Master(), "MasterActor")
  context.watch(masterRegistryActor)

  masterRegistryActor ! Master.Watchlist("TSLA")
  masterRegistryActor ! Master.Watchlist("NVDA")

  // 周期性调度：首次延迟100ms执行，之后每1秒执行一次
  val scheduleCancellable = context.system.scheduler.scheduleWithFixedDelay(
    initialDelay = 100.milliseconds,
    delay = 1.second,
    receiver = masterRegistryActor,
    message = Master.Watchlist("AAPL")
  )(context.executionContext)

  // 监听Actor终止信号，停止调度避免内存泄漏
  Behaviors.receiveSignal[Nothing] {
    case (_, Terminated(`masterRegistryActor`)) =>
      scheduleCancellable.cancel()
      Behaviors.stopped
  }
}

问题2 全局获取masterRegistryActor引用的方案 #

推荐使用Akka Typed官方提供的Receptionist服务发现机制实现，无需硬编码路径，支持分布式场景，实现步骤如下：

1. 首先在Master的伴生对象中定义服务注册Key：

import akka.actor.typed.receptionist.ServiceKey

object Master {
  // 定义服务Key，泛型为Master Actor接收的消息类型
  val MasterServiceKey: ServiceKey[Command] = ServiceKey[Command]("master-registry-actor")
  // 你的原有消息定义不变
  sealed trait Command
  case class Watchlist(code: String) extends Command
  // ...其余消息定义
}

2. 在Behaviors.setup中注册Actor到Receptionist：

val rootBehavior = Behaviors.setup[Nothing] { context =>
  val masterRegistryActor = context.spawn(Master(), "MasterActor")
  // 注册到全局服务发现
  context.system.receptionist ! Receptionist.Register(Master.MasterServiceKey, masterRegistryActor)
  // 其余原有逻辑不变
  // ...
}

3. 应用任意位置只要持有ActorSystem实例，即可查询到该Actor引用：

// 示例：在外部通过system查询Master Actor
import akka.actor.typed.receptionist.Receptionist

system.receptionist ! Receptionist.Find(Master.MasterServiceKey, replyToActor)
// 也可以通过ask模式异步获取引用
import akka.actor.typed.scaladsl.AskPattern._
import scala.concurrent.Future

implicit val timeout: akka.util.Timeout = 3.seconds
val masterRefFuture: Future[Receptionist.Listing] = system.ask(Receptionist.Find(Master.MasterServiceKey))
masterRefFuture.foreach { listing =>
  val masterRef = listing.serviceInstances(Master.MasterServiceKey).head
  // 拿到引用后即可发消息
  masterRef ! Master.Watchlist("MSFT")
}

如果是单实例小型应用，也可以用全局变量存储引用，注意保证初始化顺序即可，该方案不支持分布式场景。 #

内容的提问来源于stack exchange，提问作者Blankman