Alarms

背景

Durable Objects alarm 允许您调度 Durable Object 在未来某个时间被唤醒。当 alarm 的调度时间到达时，将调用 alarm() 处理器方法。Alarm 使用 Storage API 修改，alarm 操作遵循与其他存储操作相同的规则。

值得注意的是：

每个 Durable Object 通过调用 setAlarm() 一次只能调度一个 alarm。
Alarm 保证至少一次执行，当 alarm() 处理器抛出异常时会自动重试。
重试使用指数退避，从首次失败起延迟 2 秒，最多允许 6 次重试。

Alarm 可用于构建分布式原语，如基于 Durable Objects 的 Queues 或工作批处理。Alarm 还提供一种机制，保证 Durable Object 内的操作完成，而无需依赖传入请求来保持 Durable Object 存活。完整示例请参阅使用 Alarms API。

使用单个 alarm 调度多个事件

虽然每个 Durable Object 一次只能设置一个 alarm，但您可以通过将事件调度存储在 storage 中，让 alarm() 处理器处理到期事件，然后为下一个事件重新调度自身，来管理许多调度和 recurring 事件。

import { DurableObject } from "cloudflare:workers";

export class AgentServer extends DurableObject {
  // Schedule a one-time or recurring event
  async scheduleEvent(id, runAt, repeatMs = null) {
    await this.ctx.storage.put(`event:${id}`, { id, runAt, repeatMs });
    const currentAlarm = await this.ctx.storage.getAlarm();
    if (!currentAlarm || runAt < currentAlarm) {
      await this.ctx.storage.setAlarm(runAt);
    }
  }

  async alarm() {
    const now = Date.now();
    const events = await this.ctx.storage.list({ prefix: "event:" });
    let nextAlarm = null;

    for (const [key, event] of events) {
      if (event.runAt <= now) {
        await this.processEvent(event);
        if (event.repeatMs) {
          event.runAt = now + event.repeatMs;
          await this.ctx.storage.put(key, event);
        } else {
          await this.ctx.storage.delete(key);
        }
      }
      // Track the next event time
      if (event.runAt > now && (!nextAlarm || event.runAt < nextAlarm)) {
        nextAlarm = event.runAt;
      }
    }

    if (nextAlarm) await this.ctx.storage.setAlarm(nextAlarm);
  }

  async processEvent(event) {
    // Your event handling logic here
  }
}

Storage 方法

`getAlarm`

getAlarm(): number | null
- 如果设置了 alarm，则返回当前设置的 alarm 时间，以自 UNIX 纪元以来经过的毫秒数表示。否则返回 null。
- 如果在 alarm 已在运行时调用 getAlarm，则返回 null，除非自 alarm 处理器开始运行以来也调用了 setAlarm。

`setAlarm`

setAlarm(scheduledTimeMs number) : void
- 设置 alarm 运行的时间。将时间指定为自 UNIX 纪元以来经过的毫秒数。
- 如果已有 alarm 调度时调用 setAlarm，将覆盖现有 alarm。

`deleteAlarm`

deleteAlarm(): void
- 如果当前设置了 alarm，则取消 alarm。
- 在 alarm() 处理器内调用 deleteAlarm() 可能会尽力阻止重试，但不保证。

处理器方法

`alarm`

alarm(alarmInfo Object): void
- 当调度的 alarm 时间到达时由系统调用。
- 可选参数 alarmInfo 对象有两个属性：
  - retryCount number: 此 alarm 事件已重试的次数。
  - isRetry boolean: 布尔值，指示 alarm 是否已重试。如果此 alarm 事件是重试，则此值为 true。
- 每个 Durable Object 实例在任意给定时刻只会运行一个 alarm() 实例。
- alarm() 处理器保证至少一次执行，失败时使用指数退避重试，从 2 秒延迟开始，最多 6 次重试。这仅适用于最近的 setAlarm() 调用。如果方法因未捕获的异常而失败，将执行重试。
- 此方法可以是 async。

示例

此示例展示如何使用 setAlarm(timestamp) 方法设置 alarm，以及在 Durable Object 内使用 alarm() 处理器处理 alarm。

每次 alarm 触发时都会调用 alarm() 处理器。
如果意外错误终止 Durable Object，alarm() 处理器可能在另一台机器上重新实例化。
短暂延迟后，alarm() 处理器将在另一台机器上从头运行。

import { DurableObject } from "cloudflare:workers";

export default {
	async fetch(request, env) {
		return await env.ALARM_EXAMPLE.getByName("foo").fetch(request);
	},
};

const SECONDS = 1000;

export class AlarmExample extends DurableObject {
	constructor(ctx, env) {
		super(ctx, env);
		this.storage = ctx.storage;
	}
	async fetch(request) {
		// If there is no alarm currently set, set one for 10 seconds from now
		let currentAlarm = await this.storage.getAlarm();
		if (currentAlarm == null) {
			this.storage.setAlarm(Date.now() + 10 * SECONDS);
		}
	}
	async alarm() {
		// The alarm handler will be invoked whenever an alarm fires.
		// You can use this to do work, read from the Storage API, make HTTP calls
		// and set future alarms to run using this.storage.setAlarm() from within this handler.
	}
}

import time

from workers import DurableObject, WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return await self.env.ALARM_EXAMPLE.getByName("foo").fetch(request)

SECONDS = 1000

class AlarmExample(DurableObject):
    def __init__(self, ctx, env):
        super().__init__(ctx, env)
        self.storage = ctx.storage

    async def fetch(self, request):
        # If there is no alarm currently set, set one for 10 seconds from now
        current_alarm = await self.storage.getAlarm()
        if current_alarm is None:
            self.storage.setAlarm(int(time.time() * 1000) + 10 * SECONDS)

    async def alarm(self):
        # The alarm handler will be invoked whenever an alarm fires.
        # You can use this to do work, read from the Storage API, make HTTP calls
        # and set future alarms to run using self.storage.setAlarm() from within this handler.
        pass

以下示例展示如何使用 alarmInfo 属性识别 alarm 事件是否之前已尝试过。

class MyDurableObject extends DurableObject {
	async alarm(alarmInfo) {
		if (alarmInfo?.retryCount != 0) {
			console.log(
				"This alarm event has been attempted ${alarmInfo?.retryCount} times before.",
			);
		}
	}
}

class MyDurableObject(DurableObject):
    async def alarm(self, alarm_info):
        if alarm_info and alarm_info.get('retryCount', 0) != 0:
            print(f"This alarm event has been attempted {alarm_info.get('retryCount')} times before.")

Alarms

背景

使用单个 alarm 调度多个事件

Storage 方法

getAlarm

setAlarm

deleteAlarm

处理器方法

alarm

示例

相关资源

`getAlarm`

`setAlarm`

`deleteAlarm`

`alarm`