Akka定时任务schedule()方法

文章目录

• Akka定时任务schedule()方法
• • 什么是Akka定时任务schedule()方法？
  • 如何使用Akka定时任务schedule()方法？
  • • 如何在actor外部获取Scheduler对象
    • • 为什么需要提供一个隐式的ExecutionContext对象，用于执行定时任务？
    • 如何在actor内部获取Scheduler对象
    • schedule()方法的格式
  • Akka定时任务schedule()方法有哪些类型的延迟？
  • • 固定延迟
    • 固定频率
  • Duration类
  • 总结

什么是Akka定时任务schedule()方法？

Akka定时任务schedule()方法是一种在Akka actor系统中管理周期性执行任务的方式。它可以让我们在指定的时间点或时间间隔，向actor发送消息或执行函数。它返回一个Cancellable对象，我们可以调用它的cancel()方法来取消定时任务的执行。

Akka定时任务schedule()方法是基于一个哈希轮定时器（Hashed Wheel Timer）实现的，它是一种高效的数据结构和算法，用于处理大量的定时触发事件，如actor接收超时、Future超时、断路器等。它不会精确地在指定的时间点执行任务，而是在每个滴答（tick）时，执行所有已经到期的任务。我们可以通过配置属性akka.scheduler.tick-duration来修改Akka定时任务schedule()方法的精度。

如何使用Akka定时任务schedule()方法？

如何在actor外部获取Scheduler对象

要使用Akka定时任务schedule()方法，我们需要先获取一个Scheduler对象，它是每个ActorSystem唯一的，并且用于内部调度各种事件。我们可以通过调用ActorSystem的scheduler方法来获取它：

// 获取一个ActorSystem对象
val system = ActorSystem("akka-scheduler-system")
// 获取一个Scheduler对象
val scheduler = system.scheduler

然后，我们需要提供一个隐式的ExecutionContext对象，用于执行定时任务。我们可以使用ActorSystem自带的dispatcher作为ExecutionContext：

// 导入ActorSystem自带的dispatcher作为ExecutionContext
import system.dispatcher

为什么需要提供一个隐式的ExecutionContext对象，用于执行定时任务？

ExecutionContext是一个表示执行上下文的特质，它可以异步地执行程序逻辑，通常但不一定是在一个线程池上。它类似于Java的Executor接口，它有两个抽象方法：execute和reportFailure。execute方法用于执行一个Runnable对象，reportFailure方法用于报告一个异步计算失败的原因。

当我们使用Akka定时任务schedule()方法创建定时任务时，我们需要提供一个ExecutionContext对象，用于执行我们传入的函数或Runnable对象。这样，我们可以控制定时任务的执行策略，例如使用哪个线程池，如何处理异常等。我们可以使用ActorSystem自带的dispatcher作为ExecutionContext，也可以自定义一个ExecutionContext。

为了方便使用，我们通常会把ExecutionContext对象作为一个隐式参数传入schedule()方法。这样，我们就不需要显式地指定ExecutionContext对象，只需要在作用域内导入或定义一个隐式的ExecutionContext对象即可。例如：

// 导入ActorSystem自带的dispatcher作为隐式的ExecutionContext
import system.dispatcher
// 创建一个定时任务，不需要显式地传入ExecutionContext
scheduler.schedule(100.millis, 1.second)(() => greeter ! greet)

或者

// 定义一个自定义的ExecutionContext作为隐式的ExecutionContext
implicit val ec: ExecutionContext = ...
// 创建一个定时任务，不需要显式地传入ExecutionContext
scheduler.schedule(100.millis, 1.second)(() => greeter ! greet)

这样，我们就可以简化代码的书写和理解，同时保留了定时任务执行策略的灵活性。

如果我们不想使用隐式参数的方式来传入ExecutionContext对象，我们可以显式地指定ExecutionContext对象，只需要在schedule()方法的最后一个参数列表中加上ExecutionContext对象即可。例如：

// 定义一个自定义的ExecutionContext对象
val ec: ExecutionContext = ...
// 创建一个定时任务，显式地传入ExecutionContext对象
scheduler.schedule(100.millis, 1.second)(() => greeter ! greet)(ec)

这样，我们就可以显式地指定ExecutionContext对象，而不依赖于作用域内的隐式参数。这种写法可能会更清晰和安全，但也会增加代码的冗余和复杂度。因此，我们应该根据具体的场景和需求来选择是否使用隐式或显式的方式来传入ExecutionContext对象。

如何在actor内部获取Scheduler对象

context是一个ActorContext对象，它表示一个actor的上下文信息，包括它的self引用，它的子actor，它的监督策略等。context也有一个system属性，它是一个ActorSystem对象，表示这个actor所属的actor系统。因此，context.system.scheduler.schedule实际上就是调用这个actor所属的actor系统的scheduler方法来创建定时任务。

这种写法通常出现在actor内部，当我们想要在actor内部创建定时任务时，我们可以使用context.system.scheduler.schedule来获取Scheduler对象。例如：

// 在actor内部创建一个定时任务，向自己发送HeartBeat消息
context.system.scheduler.schedule(100.millis, 1.second)(() => self ! HeartBeat)

这样，我们就不需要显式地获取或传入ActorSystem对象，只需要使用context.system即可。

schedule()方法的格式

接下来，我们可以使用Scheduler对象的schedule()方法来创建单次执行或重复执行的定时任务。这个方法有两种重载形式：

• def schedule(initialDelay: FiniteDuration, interval: FiniteDuration)(f: ⇒ Unit): Cancellable：这个方法用于创建重复执行的定时任务，它接受三个参数：初始延迟（initialDelay），表示第一次执行任务的延迟时间；间隔时间（interval），表示每次执行任务之间的时间间隔；函数（f），表示要执行的任务。它返回一个Cancellable对象，用于取消定时任务。
• def schedule(initialDelay: FiniteDuration, interval: FiniteDuration, receiver: ActorRef, message: Any): Cancellable：这个方法用于创建重复向actor发送消息的定时任务，它接受四个参数：初始延迟（initialDelay），表示第一次发送消息的延迟时间；间隔时间（interval），表示每次发送消息之间的时间间隔；接收者（receiver），表示要发送消息给哪个actor；消息（message），表示要发送什么消息。它返回一个Cancellable对象，用于取消定时任务。

如果我们只想创建单次执行或单次发送消息的定时任务，我们可以将间隔时间设置为Duration.Zero。

Akka定时任务schedule()方法有哪些类型的延迟？

Akka定时任务schedule()方法有两种类型的延迟：固定延迟（fixed-delay）和固定频率（fixed-rate）。

固定延迟

固定延迟指的是两次连续的任务执行之间的延迟时间总是至少等于给定的间隔时间。下一次任务执行的时间只有在当前任务执行完成后才会计算。如果当前任务是一个长时间运行的任务，那么下一次任务执行会延迟。因此，两次连续的任务执行之间的时间差可能不是恒定的。

例如：

// 创建一个重复向actor发送消息的定时任务，初始延迟为100毫秒，间隔时间为1秒
scheduler.schedule(100.millis, 1.second, greeter, greet)

或者

// 创建一个重复执行函数的定时任务，初始延迟为10毫秒，间隔时间为250毫秒
scheduler.schedule(10.millis, 250.millis)(() => greeter ! greet)

注意，schedule()方法只能创建固定延迟的定时任务，如果我们想创建固定频率的定时任务，我们需要使用新的scheduleAtFixedRate()方法。

从Akka 2.6版本开始，schedule()方法已经被弃用，Akka建议我们使用scheduleWithFixedDelay()方法来创建固定延迟的定时任务。所以，我们可以重写前面的例子，使用scheduleWithFixedDelay()方法：

// 创建一个重复向actor发送消息的定时任务，初始延迟为100毫秒，间隔时间为1秒
scheduler.scheduleWithFixedDelay(100.millis, 1.second, greeter, greet)

或者

// 创建一个重复执行函数的定时任务，初始延迟为10毫秒，间隔时间为250毫秒
scheduler.scheduleWithFixedDelay(10.millis, 250.millis)(() => greeter ! greet)

固定频率

固定频率指的是两次连续的任务执行之间的时间差总是等于给定的间隔时间。下一次任务执行的时间是根据上一次任务执行的时间和间隔时间来计算的。如果当前任务是一个长时间运行的任务，那么下一次任务执行可能会立即开始，或者被跳过。因此，两次连续的任务执行之间的时间差可能是恒定的，也可能是零。

从Akka 2.6版本开始，Akka建议我们使用scheduleAtFixedRate()方法来创建固定频率的定时任务。所以，我们可以重写前面的例子，使用scheduleAtFixedRate()方法：

// 创建一个重复向actor发送消息的定时任务，初始延迟为100毫秒，间隔时间为1秒
scheduler.scheduleAtFixedRate(100.millis, 1.second, greeter, greet)

或者

// 创建一个重复执行函数的定时任务，初始延迟为10毫秒，间隔时间为250毫秒
scheduler.scheduleAtFixedRate(10.millis, 250.millis)(() => greeter ! greet)

Duration类

在上一篇利用akka模拟Spark的Master与Worker通信中，为什么那篇文章里的参数里面是Duration(3,TimeUnit.SECONDS)，而不是这篇提供的这种。

Duration是一个表示时间长度的类，它有多种构造方法，可以接受不同类型的参数。这篇提供的那种参数是使用了一个隐式转换，把一个数字和一个时间单位组合成一个Duration对象。例如：

// 使用隐式转换，把3秒转换为一个Duration对象
3.seconds
// 等价于
Duration(3, TimeUnit.SECONDS)

这种隐式转换是由scala.concurrent.duration包提供的，它可以让我们用更简洁和直观的方式来表示时间长度。我们需要导入这个包才能使用这种隐式转换：

// 导入scala.concurrent.duration包
import scala.concurrent.duration._

所以参数里面是Duration(3,TimeUnit.SECONDS)或者3.seconds都是可以的。

参数里面是Duration(3,TimeUnit.SECONDS)可能是因为没有导入这个包，或者是为了避免歧义，或者是为了保持一致性。它们的效果是一样的，只是写法不同而已。

总结

在这篇文章中，介绍了Akka定时任务schedule()方法，它是一种在Akka actor系统中管理周期性执行任务的方式。看到了如何使用schedule()方法来创建单次执行和重复执行的定时任务，以及它们的不同类型的延迟。还了解了Akka定时任务schedule()方法的实现原理和精度问题。