java - 非阻塞定时器任务和 Akka Actor

Question

请注意:虽然我更喜欢使用 Akka 的 Java API(这就是我正在使用的)的解决方案，但我对任何可行的解决方案都很满意也许可以弄清楚如何将基于 Scala 的答案转换为 Java 语言。

<小时/>

我有一个 Akka 应用程序，它有很多 actor，其中两个是 Fizz 和 Buzz。 Fizz actor 可以接受 2 种类型的消息:

• StartNewTimerTask;和
• 重置所有定时器任务

Buzz actor 只是接受一条 DoItNow 消息。这些参与者与系统其余部分之间的消息流如下:

1. 任何东西(其他 actor，甚至是 actor 系统外部的事件驱动组件)都可以随时向 Fizz actor 发送 StartNewTimerTask 消息
2. 每次 Fizz actor 收到 StartNewTimerTask 消息时，它都会创建并启动一个新的异步/非阻塞计时器，该计时器尝试运行比如说 8 秒。如果计时器到达终点(8 秒)，则会向 Buzz actor 发送一条 DoItNow 消息
3. Fizz actor 可以接受任意数量的并发 StartNewTimerTask 消息，因此，可以同时“管理”多个计时器，每个计时器都在计数朝着那个 8 秒的神奇数字前进。因此，如果其他 20 个参与者在几秒钟内向 Fizz actor 发送 StartNewTimerTask 消息，那么 Fizz actor 将“管理“同时有20个非阻塞、独立的定时器。当这 20 个计时器中的每一个达到各自的 8 秒持续时间时，它们都会向 Buzz actor 发送 20 条独立的 DoItNow 消息
4. 当 Fizz actor 收到 ResetAllTimerTasks 消息时，任何当前“正在进行”的计时器都将被中断/取消(以便他们停止倒计时 8 秒持续时间，从而阻止他们向 Buzz 发送 DoItNow 消息。因此，借用上面的示例，如果在 t=1 和 t=3 之间，Fizz actor 收到 20 个 StartNewTimerTask 消息，那么在 t=10 时，可能有 14 个各自的计时器已过去并触发了 DoItNow 消息，并且可能有 6 个计时器仍在进行中。如果在那一刻 Fizz 收到 ResetAllTimerTasks 消息，它将阻止这 6 个计时器消逝并触发消息，因此在本例中 Buzz只会收到 14 条 DoItNow 消息

我知道 Java 8 API(无 Akka)提倡扩展 TimerTask 并将这些任务提交给 Timer#scheduleAtFixedRate 方法，但我不确定是否这样做与 Akka 完全冲突，或者是否有更好的方法使用 Akka API 实现此功能。迄今为止我最好的尝试:

// Groovy pseudo-code
class MyTimerTask extends TimerTask {
  @Inject
  ActorRef buzz

  @Override
  void run() {
    // No op!
  }

  void completeTask() {
    buzz.tell(new DoItNow(), null)
  }
}

class Fizz extends UntypedAbstractActor {
  @Inject
  Timer timer

  @Override
  void onReceive(Object message) {
    if(message in StartNewTimerTask) {
      timer.scheduleAtFixedRate(new MyTimerTask(), 0, 8 * 1000)
    } else if(message in ResetAllTimerTasks) {
      timer.cancel()
    }
  }
}

class Buzz extends UntypedAbstractActor {
  @Override
  void onReceive(Object message) {
    if(message in DoItNow) {
      // Do something super cool now...
    }
  }
}

但是，我认为我没有正确管理计时器，也没有充分发挥 Akka 调度程序/计时器 API 的潜力。 有什么想法吗？

Answer 1

考虑避开 Java Timer API支持新的actor timer刚刚随 Akka 2.5.4 发布的功能。 Actor 计时器允许 Actor 使用与其生命周期相关的一个或多个内部计时器来为其自身安排定期消息。要在 Java 中访问此功能，只需更改 Fizz Actor 延长 AbstractActorWithTimers .

下面的示例是 Scala 中的(在 Scala 中，混合 Timers 特征):

object Fizz {
  private case object SendToBuzz
}

class Fizz(buzz: ActorRef) extends Actor with Timers {
  import Fizz._

  def receive = {
    case StartNewTimerTask =>
      val uuid = java.util.UUID.randomUUID
      timers.startPeriodicTimer(uuid, SendToBuzz, 8.seconds)
    case ResetAllTimerTasks =>
      timers.cancelAll()
    case SendToBuzz =>
      buzz ! DoItNow
  }
}

• 当 Fizz actor 处理 StartNewTimerTask消息，它启动一个新的计时器，该计时器将发送 SendToBuzz留言至self (即 Fizz Actor )每八秒一次。
• 何时 Fizz处理 SendToBuzz消息，它发送 DoItNow留言至Buzz Actor 。
• 何时 Fizz处理 ResetAllTimerTasks消息，它取消所有计时器。
• 如果Fizz重新启动或停止时，其所有计时器都会自动取消。