Durable Objects 为构建有状态、协调的应用提供了强大的原语。每个 Durable Object 是具有独立持久存储的单线程、全球唯一实例。围绕这些属性进行设计对于构建有效应用至关重要。
这是关于如何构建更有效、更正确的 Durable Object 应用的指南。
Workers 是无状态函数:每个请求可能在不同实例、不同位置运行,请求之间没有共享内存。Durable Objects 是有状态计算:每个实例具有唯一身份,在单一位置运行,并在请求之间保持状态。
在以下情况使用 Durable Objects:
- 协调 — 多个客户端需要与共享状态交互(聊天室、多人游戏、协作文档)
- 强一致性 — 操作必须串行化以避免竞态条件(库存管理、预订系统、回合制游戏)
- 按实体存储 — 每个用户、租户或资源需要独立的隔离数据库(多租户 SaaS、每用户数据)
- 持久连接 — 跨请求存活的长连接 WebSocket(实时通知、实时更新)
- 每实体调度工作 — 每个实体需要自己的计时器或调度任务(订阅续期、游戏超时)
在以下情况使用普通 Workers:
- 无状态请求处理 — 无共享状态的 API 端点、代理或转换
- 最大全球分布 — 请求应在最近的边缘位置处理
- 高扇出 — 每个请求独立且可并行处理
import { DurableObject } from "cloudflare:workers";
// ✅ Good use of Durable Objects: Seat booking requires coordination
// All booking requests for a venue must be serialized to prevent double-booking
export class SeatBooking extends DurableObject {
async bookSeat(seatId, userId) {
// Check if seat is already booked
const existing = this.ctx.storage.sql
.exec("SELECT user_id FROM bookings WHERE seat_id = ?", seatId)
.toArray();
if (existing.length > 0) {
return { success: false, message: "Seat already booked" };
}
// Book the seat - this is safe because Durable Objects are single-threaded
this.ctx.storage.sql.exec(
"INSERT INTO bookings (seat_id, user_id, booked_at) VALUES (?, ?, ?)",
seatId,
userId,
Date.now(),
);
return { success: true, message: "Seat booked successfully" };
}
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
const eventId = url.searchParams.get("event") ?? "default";
// Route to a Durable Object by event ID
// All bookings for the same event go to the same instance
const id = env.BOOKING.idFromName(eventId);
const booking = env.BOOKING.get(id);
const { seatId, userId } = await request.json();
const result = await booking.bookSeat(seatId, userId);
return Response.json(result, {
status: result.success ? 200 : 409,
});
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
BOOKING: DurableObjectNamespace<SeatBooking>;
}
// ✅ Good use of Durable Objects: Seat booking requires coordination
// All booking requests for a venue must be serialized to prevent double-booking
export class SeatBooking extends DurableObject<Env> {
async bookSeat(
seatId: string,
userId: string
): Promise<{ success: boolean; message: string }> {
// Check if seat is already booked
const existing = this.ctx.storage.sql
.exec<{ user_id: string }>(
"SELECT user_id FROM bookings WHERE seat_id = ?",
seatId
)
.toArray();
if (existing.length > 0) {
return { success: false, message: "Seat already booked" };
}
// Book the seat - this is safe because Durable Objects are single-threaded
this.ctx.storage.sql.exec(
"INSERT INTO bookings (seat_id, user_id, booked_at) VALUES (?, ?, ?)",
seatId,
userId,
Date.now()
);
return { success: true, message: "Seat booked successfully" };
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url);
const eventId = url.searchParams.get("event") ?? "default";
// Route to a Durable Object by event ID
// All bookings for the same event go to the same instance
const id = env.BOOKING.idFromName(eventId);
const booking = env.BOOKING.get(id);
const { seatId, userId } = await request.json<{
seatId: string;
userId: string;
}>();
const result = await booking.bookSeat(seatId, userId);
return Response.json(result, {
status: result.success ? 200 : 409,
});
},
};常见模式是使用 Workers 作为无状态入口点,在需要协调时将请求路由到 Durable Objects。Worker 处理身份验证、验证和响应格式化,而 Durable Object 处理有状态逻辑。
最重要的设计决策是选择每个 Durable Object 代表什么。为每个需要协调的逻辑单元创建一个 Durable Object:聊天室、游戏会话、文档、用户数据或租户工作区。
这是使 Durable Objects 强大的关键洞察。不是使用带锁的共享数据库,应用的每个"原子"都获得自己的单线程执行环境和私有存储。
import { DurableObject } from "cloudflare:workers";
// Each chat room is its own Durable Object instance
export class ChatRoom extends DurableObject {
async sendMessage(userId, message) {
// All messages to this room are processed sequentially by this single instance.
// No race conditions, no distributed locks needed.
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
message,
Date.now(),
);
}
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
const roomId = url.searchParams.get("room") ?? "lobby";
// Each room ID maps to exactly one Durable Object instance globally
const id = env.CHAT_ROOM.idFromName(roomId);
const stub = env.CHAT_ROOM.get(id);
await stub.sendMessage("user-123", "Hello, room!");
return new Response("Message sent");
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
// Each chat room is its own Durable Object instance
export class ChatRoom extends DurableObject<Env> {
async sendMessage(userId: string, message: string) {
// All messages to this room are processed sequentially by this single instance.
// No race conditions, no distributed locks needed.
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
message,
Date.now()
);
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url);
const roomId = url.searchParams.get("room") ?? "lobby";
// Each room ID maps to exactly one Durable Object instance globally
const id = env.CHAT_ROOM.idFromName(roomId);
const stub = env.CHAT_ROOM.get(id);
await stub.sendMessage("user-123", "Hello, room!");
return new Response("Message sent");
},
};不要创建处理所有请求的单个"全局" Durable Object:
import { DurableObject } from "cloudflare:workers";
// 🔴 Bad: A single Durable Object handling ALL chat rooms
export class ChatRoom extends DurableObject {
async sendMessage(roomId, userId, message) {
// All messages for ALL rooms go through this single instance.
// This becomes a bottleneck as traffic grows.
this.ctx.storage.sql.exec(
"INSERT INTO messages (room_id, user_id, content) VALUES (?, ?, ?)",
roomId,
userId,
message,
);
}
}
export default {
async fetch(request, env) {
// 🔴 Bad: Always using the same ID means one global instance
const id = env.CHAT_ROOM.idFromName("global");
const stub = env.CHAT_ROOM.get(id);
await stub.sendMessage("room-123", "user-456", "Hello!");
return new Response("Sent");
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
// 🔴 Bad: A single Durable Object handling ALL chat rooms
export class ChatRoom extends DurableObject<Env> {
async sendMessage(roomId: string, userId: string, message: string) {
// All messages for ALL rooms go through this single instance.
// This becomes a bottleneck as traffic grows.
this.ctx.storage.sql.exec(
"INSERT INTO messages (room_id, user_id, content) VALUES (?, ?, ?)",
roomId,
userId,
message
);
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
// 🔴 Bad: Always using the same ID means one global instance
const id = env.CHAT_ROOM.idFromName("global");
const stub = env.CHAT_ROOM.get(id);
await stub.sendMessage("room-123", "user-456", "Hello!");
return new Response("Sent");
},
};单个 Durable Object 对于简单操作大约可处理 500-1,000 请求/秒。此限制因每个请求执行的工作量而异:
| 操作类型 | 吞吐量 |
|---|---|
| 简单透传(最少解析) | ~1,000 req/sec |
| 中等处理(JSON 解析、验证) | ~500-750 req/sec |
| 复杂操作(转换、存储写入) | ~200-500 req/sec |
建模"原子"时,请考虑预期请求速率。如果用例超过这些限制,请将工作负载分片到多个 Durable Objects。
例如,考虑一个有 50,000 并发玩家、每秒发送 10 次更新的实时游戏。这总共产生 500,000 请求/秒。您需要 500-1,000 个游戏会话 Durable Objects——而不是一个全局协调器。
计算分片需求:
Required DOs = (Total requests/second) / (Requests per DO capacity)对有意义、确定性的字符串使用 getByName() 以实现一致路由。相同输入始终产生相同的 Durable Object ID,确保同一逻辑实体的请求始终到达同一实例。
import { DurableObject } from "cloudflare:workers";
export class GameSession extends DurableObject {
async join(playerId) {
// Game logic here
}
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
const gameId = url.searchParams.get("game");
if (!gameId) {
return new Response("Missing game ID", { status: 400 });
}
// ✅ Good: Deterministic ID from a meaningful string
// All requests for "game-abc123" go to the same Durable Object
const stub = env.GAME_SESSION.getByName(gameId);
await stub.join("player-xyz");
return new Response("Joined game");
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
GAME_SESSION: DurableObjectNamespace<GameSession>;
}
export class GameSession extends DurableObject<Env> {
async join(playerId: string) {
// Game logic here
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url);
const gameId = url.searchParams.get("game");
if (!gameId) {
return new Response("Missing game ID", { status: 400 });
}
// ✅ Good: Deterministic ID from a meaningful string
// All requests for "game-abc123" go to the same Durable Object
const stub = env.GAME_SESSION.getByName(gameId);
await stub.join("player-xyz");
return new Response("Joined game");
},
};创建 stub 不会实例化或唤醒 Durable Object。只有在您调用 stub 上的方法时,Durable Object 才会被激活。
仅在需要新的随机实例并会在外部存储映射关系时使用 newUniqueId():
import { DurableObject } from "cloudflare:workers";
export class GameSession extends DurableObject {
async join(playerId) {
// Game logic here
}
}
export default {
async fetch(request, env) {
// newUniqueId() creates a random ID - useful when creating new instances
// You must store this ID somewhere (e.g., D1) to find it again later
const id = env.GAME_SESSION.newUniqueId();
const stub = env.GAME_SESSION.get(id);
// Store the mapping: gameCode -> id.toString()
// await env.DB.prepare("INSERT INTO games (code, do_id) VALUES (?, ?)").bind(gameCode, id.toString()).run();
return Response.json({ gameId: id.toString() });
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
GAME_SESSION: DurableObjectNamespace<GameSession>;
}
export class GameSession extends DurableObject<Env> {
async join(playerId: string) {
// Game logic here
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
// newUniqueId() creates a random ID - useful when creating new instances
// You must store this ID somewhere (e.g., D1) to find it again later
const id = env.GAME_SESSION.newUniqueId();
const stub = env.GAME_SESSION.get(id);
// Store the mapping: gameCode -> id.toString()
// await env.DB.prepare("INSERT INTO games (code, do_id) VALUES (?, ?)").bind(gameCode, id.toString()).run();
return Response.json({ gameId: id.toString() });
},
};不要将所有数据放在单个 Durable Object 中。当您有层级数据(工作区包含项目、游戏服务器管理对局)时,为每个实体创建独立的子 Durable Object。父对象负责协调并跟踪子对象,而子对象各自独立处理自己的状态。
这能实现并行:对不同子对象的操作可以并发进行,同时每个子对象保持自身的单线程一致性(详细了解此模式)。
import { DurableObject } from "cloudflare:workers";
// Parent: Coordinates matches, but doesn't store match data
export class GameServer extends DurableObject {
async createMatch(matchName) {
const matchId = crypto.randomUUID();
// Store reference to the child in parent's database
this.ctx.storage.sql.exec(
"INSERT INTO matches (id, name, created_at) VALUES (?, ?, ?)",
matchId,
matchName,
Date.now(),
);
// Initialize the child Durable Object
const childId = this.env.GAME_MATCH.idFromName(matchId);
const childStub = this.env.GAME_MATCH.get(childId);
await childStub.init(matchId, matchName);
return matchId;
}
async listMatches() {
// Parent knows about all matches without waking up each child
const cursor = this.ctx.storage.sql.exec(
"SELECT id, name FROM matches ORDER BY created_at DESC",
);
return cursor.toArray();
}
}
// Child: Handles its own game state independently
export class GameMatch extends DurableObject {
async init(matchId, matchName) {
await this.ctx.storage.put("matchId", matchId);
await this.ctx.storage.put("matchName", matchName);
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS players (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
score INTEGER DEFAULT 0
)
`);
}
async addPlayer(playerId, playerName) {
this.ctx.storage.sql.exec(
"INSERT INTO players (id, name, score) VALUES (?, ?, 0)",
playerId,
playerName,
);
}
async updateScore(playerId, score) {
this.ctx.storage.sql.exec(
"UPDATE players SET score = ? WHERE id = ?",
score,
playerId,
);
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
GAME_SERVER: DurableObjectNamespace<GameServer>;
GAME_MATCH: DurableObjectNamespace<GameMatch>;
}
// Parent: Coordinates matches, but doesn't store match data
export class GameServer extends DurableObject<Env> {
async createMatch(matchName: string): Promise<string> {
const matchId = crypto.randomUUID();
// Store reference to the child in parent's database
this.ctx.storage.sql.exec(
"INSERT INTO matches (id, name, created_at) VALUES (?, ?, ?)",
matchId,
matchName,
Date.now()
);
// Initialize the child Durable Object
const childId = this.env.GAME_MATCH.idFromName(matchId);
const childStub = this.env.GAME_MATCH.get(childId);
await childStub.init(matchId, matchName);
return matchId;
}
async listMatches(): Promise<{ id: string; name: string }[]> {
// Parent knows about all matches without waking up each child
const cursor = this.ctx.storage.sql.exec<{ id: string; name: string }>(
"SELECT id, name FROM matches ORDER BY created_at DESC"
);
return cursor.toArray();
}
}
// Child: Handles its own game state independently
export class GameMatch extends DurableObject<Env> {
async init(matchId: string, matchName: string) {
await this.ctx.storage.put("matchId", matchId);
await this.ctx.storage.put("matchName", matchName);
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS players (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
score INTEGER DEFAULT 0
)
`);
}
async addPlayer(playerId: string, playerName: string) {
this.ctx.storage.sql.exec(
"INSERT INTO players (id, name, score) VALUES (?, ?, 0)",
playerId,
playerName
);
}
async updateScore(playerId: string, score: number) {
this.ctx.storage.sql.exec(
"UPDATE players SET score = ? WHERE id = ?",
score,
playerId
);
}
}采用此模式时:
- 列出对局只查询父对象(子对象保持休眠)
- 不同对局可并行处理玩家操作
- 每个对局拥有自己的 SQLite 数据库来存储玩家数据
默认情况下,Durable Object 会在首次收到请求的位置附近创建。对大多数应用来说,这已经足够好。不过,您可以提供位置提示(location hint)以影响 Durable Object 的创建位置。
import { DurableObject } from "cloudflare:workers";
export class GameSession extends DurableObject {
// Game session logic
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
const gameId = url.searchParams.get("game") ?? "default";
const region = url.searchParams.get("region") ?? "wnam"; // Western North America
// Provide a location hint for where this Durable Object should be created
const id = env.GAME_SESSION.idFromName(gameId);
const stub = env.GAME_SESSION.get(id, { locationHint: region });
return new Response("Connected to game session");
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
GAME_SESSION: DurableObjectNamespace<GameSession>;
}
export class GameSession extends DurableObject<Env> {
// Game session logic
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url);
const gameId = url.searchParams.get("game") ?? "default";
const region = url.searchParams.get("region") ?? "wnam"; // Western North America
// Provide a location hint for where this Durable Object should be created
const id = env.GAME_SESSION.idFromName(gameId);
const stub = env.GAME_SESSION.get(id, { locationHint: region });
return new Response("Connected to game session");
},
};位置提示是建议,不是保证。有关可用区域和详情,请参阅数据位置。
SQLite 存储 是新 Durable Objects 推荐的存储后端。它提供熟悉的 SQL API 用于关系查询、索引、事务,以及比旧版键值存储支持的 Durable Objects 更好的性能。SQLite Durable Objects 还支持同步和异步版本的 KV API。
在 Wrangler 配置中配置 Durable Object 类使用 SQLite 存储:
{
"migrations": [
{ "tag": "v1", "new_sqlite_classes": ["ChatRoom"] }
]
}[[migrations]]
tag = "v1"
new_sqlite_classes = [ "ChatRoom" ]然后在 Durable Object 中使用 SQL API:
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
constructor(ctx, env) {
super(ctx, env);
// Create tables on first instantiation
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL
)
`);
}
async addMessage(userId, content) {
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now(),
);
}
async getRecentMessages(limit = 50) {
// Use type parameter for typed results
const cursor = this.ctx.storage.sql.exec(
"SELECT * FROM messages ORDER BY created_at DESC LIMIT ?",
limit,
);
return cursor.toArray();
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
type Message = {
id: number;
user_id: string;
content: string;
created_at: number;
};
export class ChatRoom extends DurableObject<Env> {
constructor(ctx: DurableObjectState, env: Env) {
super(ctx, env);
// Create tables on first instantiation
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL
)
`);
}
async addMessage(userId: string, content: string) {
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now()
);
}
async getRecentMessages(limit: number = 50): Promise<Message[]> {
// Use type parameter for typed results
const cursor = this.ctx.storage.sql.exec<Message>(
"SELECT * FROM messages ORDER BY created_at DESC LIMIT ?",
limit
);
return cursor.toArray();
}
}请参阅访问 Durable Objects 存储了解 SQL API 的更多详情。
在构造函数中使用 blockConcurrencyWhile() 运行迁移并初始化状态,确保在处理任何请求之前完成。这样可保证 schema 已就绪,并防止初始化期间出现竞态条件。
对于生产应用,请使用能自动处理版本跟踪与执行的迁移库:
durable-utils↗ — 提供SQLSchemaMigrations类,可在内存和存储中同时跟踪已执行的迁移。@cloudflare/actors存储工具 ↗ — Cloudflare Actors 框架使用的同一模式的参考实现。
如果您不想使用库,可以用 _sql_schema_migrations 表手动跟踪 schema 版本。以下示例演示了此方法:
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
constructor(ctx, env) {
super(ctx, env);
// blockConcurrencyWhile() ensures no requests are processed until this completes
ctx.blockConcurrencyWhile(async () => {
await this.migrate();
});
}
async migrate() {
// Create the migrations tracking table if it does not exist
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS _sql_schema_migrations (
id INTEGER PRIMARY KEY,
applied_at TEXT NOT NULL DEFAULT (datetime('now'))
);
`);
// Determine the current schema version
const version = this.ctx.storage.sql
.exec(
"SELECT COALESCE(MAX(id), 0) as version FROM _sql_schema_migrations",
)
.one().version;
if (version < 1) {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at);
INSERT INTO _sql_schema_migrations (id) VALUES (1);
`);
}
if (version < 2) {
// Future migration: add a new column
this.ctx.storage.sql.exec(`
ALTER TABLE messages ADD COLUMN edited_at INTEGER;
INSERT INTO _sql_schema_migrations (id) VALUES (2);
`);
}
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
constructor(ctx: DurableObjectState, env: Env) {
super(ctx, env);
// blockConcurrencyWhile() ensures no requests are processed until this completes
ctx.blockConcurrencyWhile(async () => {
await this.migrate();
});
}
private async migrate() {
// Create the migrations tracking table if it does not exist
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS _sql_schema_migrations (
id INTEGER PRIMARY KEY,
applied_at TEXT NOT NULL DEFAULT (datetime('now'))
);
`);
// Determine the current schema version
const version =
this.ctx.storage.sql
.exec<{ version: number }>(
"SELECT COALESCE(MAX(id), 0) as version FROM _sql_schema_migrations",
)
.one().version;
if (version < 1) {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at);
INSERT INTO _sql_schema_migrations (id) VALUES (1);
`);
}
if (version < 2) {
// Future migration: add a new column
this.ctx.storage.sql.exec(`
ALTER TABLE messages ADD COLUMN edited_at INTEGER;
INSERT INTO _sql_schema_migrations (id) VALUES (2);
`);
}
}
}Durable Objects 提供多层状态管理,各层特性不同:
| 类型 | 速度 | 持久性 | 适用场景 |
|---|---|---|---|
| 内存中(类属性) | 最快 | 驱逐或崩溃时丢失 | 缓存、活跃连接 |
| SQLite 存储 | 快 | 跨重启持久 | 主要数据存储 |
| 外部(R2、D1) | 可变 | 持久,可跨 DO 访问 | 大文件、共享数据 |
如果 Durable Object 因不活跃被从内存中驱逐,或因未捕获异常而崩溃,内存状态不会保留。请始终将重要状态持久化到 SQLite 存储。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
// In-memory cache - fast but NOT preserved across evictions or crashes
messageCache = null;
async getRecentMessages() {
// Return from cache if available (only valid while DO is in memory)
if (this.messageCache !== null) {
return this.messageCache;
}
// Otherwise, load from durable storage
const cursor = this.ctx.storage.sql.exec(
"SELECT * FROM messages ORDER BY created_at DESC LIMIT 100",
);
this.messageCache = cursor.toArray();
return this.messageCache;
}
async addMessage(userId, content) {
// ✅ Always persist to durable storage first
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now(),
);
// Then update the cache (if it exists)
// If the DO crashes here, the message is still saved in SQLite
this.messageCache = null; // Invalidate cache
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
type Message = {
id: number;
user_id: string;
content: string;
created_at: number;
};
export class ChatRoom extends DurableObject<Env> {
// In-memory cache - fast but NOT preserved across evictions or crashes
private messageCache: Message[] | null = null;
async getRecentMessages(): Promise<Message[]> {
// Return from cache if available (only valid while DO is in memory)
if (this.messageCache !== null) {
return this.messageCache;
}
// Otherwise, load from durable storage
const cursor = this.ctx.storage.sql.exec<Message>(
"SELECT * FROM messages ORDER BY created_at DESC LIMIT 100"
);
this.messageCache = cursor.toArray();
return this.messageCache;
}
async addMessage(userId: string, content: string) {
// ✅ Always persist to durable storage first
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now()
);
// Then update the cache (if it exists)
// If the DO crashes here, the message is still saved in SQLite
this.messageCache = null; // Invalidate cache
}
}与任何数据库一样,索引能显著提升对频繁过滤列的读取性能。代价是略多的存储空间和略慢的写入。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
constructor(ctx, env) {
super(ctx, env);
ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL
);
-- Index for queries filtering by user
CREATE INDEX IF NOT EXISTS idx_messages_user_id ON messages(user_id);
-- Index for time-based queries (recent messages)
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at);
-- Composite index for user + time queries
CREATE INDEX IF NOT EXISTS idx_messages_user_time ON messages(user_id, created_at);
`);
});
}
// This query benefits from idx_messages_user_time
async getUserMessages(userId, since) {
return this.ctx.storage.sql
.exec(
"SELECT * FROM messages WHERE user_id = ? AND created_at > ? ORDER BY created_at",
userId,
since,
)
.toArray();
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
constructor(ctx: DurableObjectState, env: Env) {
super(ctx, env);
ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL
);
-- Index for queries filtering by user
CREATE INDEX IF NOT EXISTS idx_messages_user_id ON messages(user_id);
-- Index for time-based queries (recent messages)
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at);
-- Composite index for user + time queries
CREATE INDEX IF NOT EXISTS idx_messages_user_time ON messages(user_id, created_at);
`);
});
}
// This query benefits from idx_messages_user_time
async getUserMessages(userId: string, since: number) {
return this.ctx.storage.sql
.exec(
"SELECT * FROM messages WHERE user_id = ? AND created_at > ? ORDER BY created_at",
userId,
since
)
.toArray();
}
}虽然 Durable Objects 是单线程的,但 JavaScript 的 async/await 可能在请求等待异步操作结果时允许多个请求交错执行。Cloudflare 运行时使用 input gate 和 output gate 来防止数据竞态,并默认保证正确性。
Input gate 会在同步 JavaScript 执行期间阻塞新事件(传入请求、fetch 响应)。await 异步操作(如 fetch() 或 KV 存储方法)会打开 input gate,允许其他请求交错。不过,存储操作提供了特殊保护:
import { DurableObject } from "cloudflare:workers";
export class Counter extends DurableObject {
// This code is safe due to input gates
async increment() {
// While these storage operations execute, no other requests
// can interleave - input gate blocks new events
const value = (await this.ctx.storage.get("count")) ?? 0;
await this.ctx.storage.put("count", value + 1);
return value + 1;
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
COUNTER: DurableObjectNamespace<Counter>;
}
export class Counter extends DurableObject<Env> {
// This code is safe due to input gates
async increment(): Promise<number> {
// While these storage operations execute, no other requests
// can interleave - input gate blocks new events
const value = (await this.ctx.storage.get<number>("count")) ?? 0;
await this.ctx.storage.put("count", value + 1);
return value + 1;
}
}Output gate 会暂存出站网络消息(响应、fetch 请求),直到待处理的存储写入完成。这确保客户端永远不会在数据尚未持久化时就收到确认:
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
async sendMessage(userId, content) {
// Write to storage - don't need to await for correctness
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now(),
);
// This response is held by the output gate until the write completes.
// The client only receives "Message sent" after data is safely persisted.
return "Message sent";
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
async sendMessage(userId: string, content: string): Promise<string> {
// Write to storage - don't need to await for correctness
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now()
);
// This response is held by the output gate until the write completes.
// The client only receives "Message sent" after data is safely persisted.
return "Message sent";
}
}写入合并(Write coalescing): 多个存储写入之间若没有插入 await 调用,会自动合并为单个原子隐式事务:
import { DurableObject } from "cloudflare:workers";
export class Account extends DurableObject {
async transfer(fromId, toId, amount) {
// ✅ Good: These writes are coalesced into one atomic transaction
this.ctx.storage.sql.exec(
"UPDATE accounts SET balance = balance - ? WHERE id = ?",
amount,
fromId,
);
this.ctx.storage.sql.exec(
"UPDATE accounts SET balance = balance + ? WHERE id = ?",
amount,
toId,
);
this.ctx.storage.sql.exec(
"INSERT INTO transfers (from_id, to_id, amount, created_at) VALUES (?, ?, ?, ?)",
fromId,
toId,
amount,
Date.now(),
);
// All three writes commit together atomically
}
// 🔴 Bad: await on KV operations breaks coalescing
async transferBrokenKV(fromId, toId, amount) {
const fromBalance = (await this.ctx.storage.get(`balance:${fromId}`)) ?? 0;
await this.ctx.storage.put(`balance:${fromId}`, fromBalance - amount);
// If the next write fails, the debit already committed!
const toBalance = (await this.ctx.storage.get(`balance:${toId}`)) ?? 0;
await this.ctx.storage.put(`balance:${toId}`, toBalance + amount);
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
ACCOUNT: DurableObjectNamespace<Account>;
}
export class Account extends DurableObject<Env> {
async transfer(fromId: string, toId: string, amount: number) {
// ✅ Good: These writes are coalesced into one atomic transaction
this.ctx.storage.sql.exec(
"UPDATE accounts SET balance = balance - ? WHERE id = ?",
amount,
fromId
);
this.ctx.storage.sql.exec(
"UPDATE accounts SET balance = balance + ? WHERE id = ?",
amount,
toId
);
this.ctx.storage.sql.exec(
"INSERT INTO transfers (from_id, to_id, amount, created_at) VALUES (?, ?, ?, ?)",
fromId,
toId,
amount,
Date.now()
);
// All three writes commit together atomically
}
// 🔴 Bad: await on KV operations breaks coalescing
async transferBrokenKV(fromId: string, toId: string, amount: number) {
const fromBalance = (await this.ctx.storage.get<number>(`balance:${fromId}`)) ?? 0;
await this.ctx.storage.put(`balance:${fromId}`, fromBalance - amount);
// If the next write fails, the debit already committed!
const toBalance = (await this.ctx.storage.get<number>(`balance:${toId}`)) ?? 0;
await this.ctx.storage.put(`balance:${toId}`, toBalance + amount);
}
}更多详情,请参阅 Durable Objects: Easy, Fast, Correct — Choose three ↗ 以及术语表。
Input gate 仅在存储操作期间提供保护。诸如 fetch() 或写入 R2 等非存储 I/O 会允许其他请求交错,从而可能导致竞态条件:
import { DurableObject } from "cloudflare:workers";
export class Processor extends DurableObject {
// ⚠️ Potential race condition: fetch() allows interleaving
async processItem(id) {
const item = await this.ctx.storage.get(`item:${id}`);
if (item?.status === "pending") {
// During this fetch, other requests CAN execute and modify storage
const result = await fetch("https://api.example.com/process");
// Another request may have already processed this item!
await this.ctx.storage.put(`item:${id}`, { status: "completed" });
}
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
PROCESSOR: DurableObjectNamespace<Processor>;
}
export class Processor extends DurableObject<Env> {
// ⚠️ Potential race condition: fetch() allows interleaving
async processItem(id: string) {
const item = await this.ctx.storage.get<{ status: string }>(`item:${id}`);
if (item?.status === "pending") {
// During this fetch, other requests CAN execute and modify storage
const result = await fetch("https://api.example.com/process");
// Another request may have already processed this item!
await this.ctx.storage.put(`item:${id}`, { status: "completed" });
}
}
}要处理这种情况,请使用乐观锁(检查并设置)模式:在外部调用之前读取版本号,然后在写入前验证版本号未被更改。
blockConcurrencyWhile() 方法保证在所提供的回调完成之前不处理其他事件,即使回调执行异步 I/O 也是如此。这对于必须原子执行的操作很有用,例如在构造函数中从存储初始化状态:
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
constructor(ctx, env) {
super(ctx, env);
// ✅ Good: Use blockConcurrencyWhile for one-time initialization
ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
content TEXT
)
`);
});
}
// 🔴 Bad: Don't use blockConcurrencyWhile on every request
async sendMessageSlow(content) {
await this.ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(
"INSERT INTO messages (content) VALUES (?)",
content,
);
});
// If this takes ~5ms, you're limited to ~200 requests/second
}
// ✅ Good: Let output gates handle consistency
async sendMessageFast(content) {
this.ctx.storage.sql.exec(
"INSERT INTO messages (content) VALUES (?)",
content,
);
// Output gate ensures write completes before response is sent
// Other requests can be processed concurrently
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
constructor(ctx: DurableObjectState, env: Env) {
super(ctx, env);
// ✅ Good: Use blockConcurrencyWhile for one-time initialization
ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
content TEXT
)
`);
});
}
// 🔴 Bad: Don't use blockConcurrencyWhile on every request
async sendMessageSlow(content: string) {
await this.ctx.blockConcurrencyWhile(async () => {
this.ctx.storage.sql.exec(
"INSERT INTO messages (content) VALUES (?)",
content
);
});
// If this takes ~5ms, you're limited to ~200 requests/second
}
// ✅ Good: Let output gates handle consistency
async sendMessageFast(content: string) {
this.ctx.storage.sql.exec(
"INSERT INTO messages (content) VALUES (?)",
content
);
// Output gate ensures write completes before response is sent
// Other requests can be processed concurrently
}
}由于 blockConcurrencyWhile() 会无条件地阻塞所有并发,它会显著降低吞吐量。如果每次调用大约耗时 5ms,该单个 Durable Object 大约只能处理 200 请求/秒。请将其保留用于初始化和迁移,而非常规请求处理。对于正常操作,请依赖 input/output gate 和写入合并。
对于请求处理期间的原子读-改-写操作,优先使用 transaction(),而非 blockConcurrencyWhile()。事务为存储操作提供原子性,而不会阻塞无关的并发请求。
兼容性日期为 2024-04-03 或之后的项目应使用 RPC 方法。RPC 更符合人体工学,提供更好的类型安全,并消除手动请求/响应解析。
在 Durable Object 类上定义公共方法,并从 stub 直接调用它们,同时获得完整的 TypeScript 支持:
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
// Public methods are automatically exposed as RPC endpoints
async sendMessage(userId, content) {
const createdAt = Date.now();
const result = this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?) RETURNING id",
userId,
content,
createdAt,
);
const { id } = result.one();
return { id, userId, content, createdAt };
}
async getMessages(limit = 50) {
const cursor = this.ctx.storage.sql.exec(
"SELECT * FROM messages ORDER BY created_at DESC LIMIT ?",
limit,
);
return cursor.toArray().map((row) => ({
id: row.id,
userId: row.user_id,
content: row.content,
createdAt: row.created_at,
}));
}
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
const roomId = url.searchParams.get("room") ?? "lobby";
const id = env.CHAT_ROOM.idFromName(roomId);
// stub is typed as DurableObjectStub<ChatRoom>
const stub = env.CHAT_ROOM.get(id);
if (request.method === "POST") {
const { userId, content } = await request.json();
// Direct method call with full type checking
const message = await stub.sendMessage(userId, content);
return Response.json(message);
}
// TypeScript knows getMessages() returns Promise<Message[]>
const messages = await stub.getMessages(100);
return Response.json(messages);
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
// Type parameter provides typed method calls on the stub
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
type Message = {
id: number;
userId: string;
content: string;
createdAt: number;
};
export class ChatRoom extends DurableObject<Env> {
// Public methods are automatically exposed as RPC endpoints
async sendMessage(userId: string, content: string): Promise<Message> {
const createdAt = Date.now();
const result = this.ctx.storage.sql.exec<{ id: number }>(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?) RETURNING id",
userId,
content,
createdAt
);
const { id } = result.one();
return { id, userId, content, createdAt };
}
async getMessages(limit: number = 50): Promise<Message[]> {
const cursor = this.ctx.storage.sql.exec<{
id: number;
user_id: string;
content: string;
created_at: number;
}>("SELECT * FROM messages ORDER BY created_at DESC LIMIT ?", limit);
return cursor.toArray().map((row) => ({
id: row.id,
userId: row.user_id,
content: row.content,
createdAt: row.created_at,
}));
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url);
const roomId = url.searchParams.get("room") ?? "lobby";
const id = env.CHAT_ROOM.idFromName(roomId);
// stub is typed as DurableObjectStub<ChatRoom>
const stub = env.CHAT_ROOM.get(id);
if (request.method === "POST") {
const { userId, content } = await request.json<{
userId: string;
content: string;
}>();
// Direct method call with full type checking
const message = await stub.sendMessage(userId, content);
return Response.json(message);
}
// TypeScript knows getMessages() returns Promise<Message[]>
const messages = await stub.getMessages(100);
return Response.json(messages);
},
};有关 RPC 与旧版 fetch() 处理器的更多详情,请参阅调用方法。
Durable Objects 在内部不知道自己的名称或 ID。如果您的 Durable Object 需要知道自身身份(例如,存储对自身的引用,或与相关对象通信),您必须显式初始化它。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
roomId = null;
// Call this after creating the Durable Object for the first time
async init(roomId, createdBy) {
// Check if already initialized
const existing = await this.ctx.storage.get("roomId");
if (existing) {
return; // Already initialized
}
// Store the identity
await this.ctx.storage.put("roomId", roomId);
await this.ctx.storage.put("createdBy", createdBy);
await this.ctx.storage.put("createdAt", Date.now());
// Cache in memory for this session
this.roomId = roomId;
}
async getRoomId() {
if (this.roomId) {
return this.roomId;
}
const stored = await this.ctx.storage.get("roomId");
if (!stored) {
throw new Error("ChatRoom not initialized. Call init() first.");
}
this.roomId = stored;
return stored;
}
}
export default {
async fetch(request, env) {
const url = new URL(request.url);
const roomId = url.searchParams.get("room") ?? "lobby";
const id = env.CHAT_ROOM.idFromName(roomId);
const stub = env.CHAT_ROOM.get(id);
// Initialize on first access
await stub.init(roomId, "system");
return new Response(`Room ${await stub.getRoomId()} ready`);
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
private roomId: string | null = null;
// Call this after creating the Durable Object for the first time
async init(roomId: string, createdBy: string) {
// Check if already initialized
const existing = await this.ctx.storage.get("roomId");
if (existing) {
return; // Already initialized
}
// Store the identity
await this.ctx.storage.put("roomId", roomId);
await this.ctx.storage.put("createdBy", createdBy);
await this.ctx.storage.put("createdAt", Date.now());
// Cache in memory for this session
this.roomId = roomId;
}
async getRoomId(): Promise<string> {
if (this.roomId) {
return this.roomId;
}
const stored = await this.ctx.storage.get<string>("roomId");
if (!stored) {
throw new Error("ChatRoom not initialized. Call init() first.");
}
this.roomId = stored;
return stored;
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url);
const roomId = url.searchParams.get("room") ?? "lobby";
const id = env.CHAT_ROOM.idFromName(roomId);
const stub = env.CHAT_ROOM.get(id);
// Initialize on first access
await stub.init(roomId, "system");
return new Response(`Room ${await stub.getRoomId()} ready`);
},
};在 Durable Object stub 上调用方法时,请始终使用 await。未 await 的调用会创建悬空 Promise,导致错误被吞掉、返回值丢失。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
async sendMessage(userId, content) {
const result = this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?) RETURNING id",
userId,
content,
Date.now(),
);
return result.one().id;
}
}
export default {
async fetch(request, env) {
const id = env.CHAT_ROOM.idFromName("lobby");
const stub = env.CHAT_ROOM.get(id);
// 🔴 Bad: Not awaiting the call
// The message ID is lost, and any errors are swallowed
stub.sendMessage("user-123", "Hello");
// ✅ Good: Properly awaited
const messageId = await stub.sendMessage("user-123", "Hello");
return Response.json({ messageId });
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
async sendMessage(userId: string, content: string): Promise<number> {
const result = this.ctx.storage.sql.exec<{ id: number }>(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?) RETURNING id",
userId,
content,
Date.now()
);
return result.one().id;
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const id = env.CHAT_ROOM.idFromName("lobby");
const stub = env.CHAT_ROOM.get(id);
// 🔴 Bad: Not awaiting the call
// The message ID is lost, and any errors are swallowed
stub.sendMessage("user-123", "Hello");
// ✅ Good: Properly awaited
const messageId = await stub.sendMessage("user-123", "Hello");
return Response.json({ messageId });
},
};Durable Object 中的未捕获异常可能使其处于未知状态,并可能导致运行时终止该实例。请将有风险的操作包裹在 try...catch 块中,并适当处理错误。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
async processMessage(userId, content) {
// ✅ Good: Wrap risky operations in try...catch
try {
// Validate input before processing
if (!content || content.length > 10000) {
throw new Error("Invalid message content");
}
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now(),
);
// External call that might fail
await this.notifySubscribers(content);
} catch (error) {
// Log the error for debugging
console.error("Failed to process message:", error);
// Re-throw if it's a validation error (don't retry)
if (error instanceof Error && error.message.includes("Invalid")) {
throw error;
}
// For transient errors, you might want to handle differently
throw error;
}
}
async notifySubscribers(content) {
// External notification logic
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
async processMessage(userId: string, content: string) {
// ✅ Good: Wrap risky operations in try...catch
try {
// Validate input before processing
if (!content || content.length > 10000) {
throw new Error("Invalid message content");
}
this.ctx.storage.sql.exec(
"INSERT INTO messages (user_id, content, created_at) VALUES (?, ?, ?)",
userId,
content,
Date.now()
);
// External call that might fail
await this.notifySubscribers(content);
} catch (error) {
// Log the error for debugging
console.error("Failed to process message:", error);
// Re-throw if it's a validation error (don't retry)
if (error instanceof Error && error.message.includes("Invalid")) {
throw error;
}
// For transient errors, you might want to handle differently
throw error;
}
}
private async notifySubscribers(content: string) {
// External notification logic
}
}从 Worker 调用 Durable Objects 时,错误可能包含 .retryable 和 .overloaded 属性,指示该操作是否可重试。对于瞬时故障,请实现指数退避,以避免压垮系统。
有关错误属性、重试策略和指数退避模式的详情,请参阅错误处理。
可休眠 WebSocket API(Hibernatable WebSockets API)允许 Durable Objects 在保持 WebSocket 连接的同时进入休眠。这对拥有大量空闲连接的应用可显著降低成本。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
async fetch(request) {
const url = new URL(request.url);
if (url.pathname === "/websocket") {
// Check for WebSocket upgrade
if (request.headers.get("Upgrade") !== "websocket") {
return new Response("Expected WebSocket", { status: 400 });
}
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
// Accept the WebSocket with Hibernation API
this.ctx.acceptWebSocket(server);
return new Response(null, { status: 101, webSocket: client });
}
return new Response("Not found", { status: 404 });
}
// Called when a message is received (even after hibernation)
async webSocketMessage(ws, message) {
const data = typeof message === "string" ? message : "binary data";
// Broadcast to all connected clients
for (const client of this.ctx.getWebSockets()) {
if (client !== ws && client.readyState === WebSocket.OPEN) {
client.send(data);
}
}
}
// Called when a WebSocket is closed
async webSocketClose(ws, code, reason, wasClean) {
// With web_socket_auto_reply_to_close (compat date >= 2026-04-07), the runtime
// auto-replies to Close frames. Calling close() is safe but no longer required.
ws.close(code, reason);
console.log(`WebSocket closed: ${code} ${reason}`);
}
// Called when a WebSocket error occurs
async webSocketError(ws, error) {
console.error("WebSocket error:", error);
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
async fetch(request: Request): Promise<Response> {
const url = new URL(request.url);
if (url.pathname === "/websocket") {
// Check for WebSocket upgrade
if (request.headers.get("Upgrade") !== "websocket") {
return new Response("Expected WebSocket", { status: 400 });
}
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
// Accept the WebSocket with Hibernation API
this.ctx.acceptWebSocket(server);
return new Response(null, { status: 101, webSocket: client });
}
return new Response("Not found", { status: 404 });
}
// Called when a message is received (even after hibernation)
async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
const data = typeof message === "string" ? message : "binary data";
// Broadcast to all connected clients
for (const client of this.ctx.getWebSockets()) {
if (client !== ws && client.readyState === WebSocket.OPEN) {
client.send(data);
}
}
}
// Called when a WebSocket is closed
async webSocketClose(
ws: WebSocket,
code: number,
reason: string,
wasClean: boolean
) {
// With web_socket_auto_reply_to_close (compat date >= 2026-04-07), the runtime
// auto-replies to Close frames. Calling close() is safe but no longer required.
ws.close(code, reason);
console.log(`WebSocket closed: ${code} ${reason}`);
}
// Called when a WebSocket error occurs
async webSocketError(ws: WebSocket, error: unknown) {
console.error("WebSocket error:", error);
}
}使用 Hibernation API 时,Durable Object 可以在没有活跃 JavaScript 执行时进入休眠,但 WebSocket 连接保持打开。当消息到达时,Durable Object 会自动唤醒。
最佳实践:
- WebSocket Hibernation API 暴露
webSocketError、webSocketMessage和webSocketClose处理器,分别对应各类 WebSocket 事件。 - 启用
web_socket_auto_reply_to_close兼容性标志后(兼容性日期为2026-04-07或之后默认启用),运行时会自动完成关闭握手。在webSocketClose中调用ws.close()仍然安全,但不再必需。在较旧的兼容性日期上,您必须调用ws.close(),以避免1006异常关闭错误。
更多详情请参阅 WebSockets。
WebSocket attachment 可让您为每个连接存储可在休眠后保留的元数据。可用于用户 ID、会话令牌或其他每连接数据。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
async fetch(request) {
const url = new URL(request.url);
if (url.pathname === "/websocket") {
if (request.headers.get("Upgrade") !== "websocket") {
return new Response("Expected WebSocket", { status: 400 });
}
const userId = url.searchParams.get("userId") ?? "anonymous";
const username = url.searchParams.get("username") ?? "Anonymous";
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
this.ctx.acceptWebSocket(server);
// Store per-connection state that survives hibernation
const state = {
userId,
username,
joinedAt: Date.now(),
};
server.serializeAttachment(state);
// Broadcast join message
this.broadcast(`${username} joined the chat`);
return new Response(null, { status: 101, webSocket: client });
}
return new Response("Not found", { status: 404 });
}
async webSocketMessage(ws, message) {
// Retrieve the connection state (works even after hibernation)
const state = ws.deserializeAttachment();
const chatMessage = JSON.stringify({
userId: state.userId,
username: state.username,
content: message,
timestamp: Date.now(),
});
this.broadcast(chatMessage);
}
async webSocketClose(ws, code, reason) {
// With web_socket_auto_reply_to_close (compat date >= 2026-04-07), the runtime
// auto-replies to Close frames. Calling close() is safe but no longer required.
ws.close(code, reason);
const state = ws.deserializeAttachment();
this.broadcast(`${state.username} left the chat`);
}
broadcast(message) {
for (const client of this.ctx.getWebSockets()) {
if (client.readyState === WebSocket.OPEN) {
client.send(message);
}
}
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
type ConnectionState = {
userId: string;
username: string;
joinedAt: number;
};
export class ChatRoom extends DurableObject<Env> {
async fetch(request: Request): Promise<Response> {
const url = new URL(request.url);
if (url.pathname === "/websocket") {
if (request.headers.get("Upgrade") !== "websocket") {
return new Response("Expected WebSocket", { status: 400 });
}
const userId = url.searchParams.get("userId") ?? "anonymous";
const username = url.searchParams.get("username") ?? "Anonymous";
const pair = new WebSocketPair();
const [client, server] = Object.values(pair);
this.ctx.acceptWebSocket(server);
// Store per-connection state that survives hibernation
const state: ConnectionState = {
userId,
username,
joinedAt: Date.now(),
};
server.serializeAttachment(state);
// Broadcast join message
this.broadcast(`${username} joined the chat`);
return new Response(null, { status: 101, webSocket: client });
}
return new Response("Not found", { status: 404 });
}
async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
// Retrieve the connection state (works even after hibernation)
const state = ws.deserializeAttachment() as ConnectionState;
const chatMessage = JSON.stringify({
userId: state.userId,
username: state.username,
content: message,
timestamp: Date.now(),
});
this.broadcast(chatMessage);
}
async webSocketClose(ws: WebSocket, code: number, reason: string) {
// With web_socket_auto_reply_to_close (compat date >= 2026-04-07), the runtime
// auto-replies to Close frames. Calling close() is safe but no longer required.
ws.close(code, reason);
const state = ws.deserializeAttachment() as ConnectionState;
this.broadcast(`${state.username} left the chat`);
}
private broadcast(message: string) {
for (const client of this.ctx.getWebSockets()) {
if (client.readyState === WebSocket.OPEN) {
client.send(message);
}
}
}
}每个 Durable Object 都可以使用 Alarms API 调度自己的未来工作,从而在没有传入请求、RPC 调用或 WebSocket 消息的情况下,按任意间隔执行后台任务。
关于 alarm 的要点:
setAlarm(timestamp)调度alarm()处理器在未来任意时间运行(毫秒精度)- Alarm 不会自动重复 — 您必须再次调用
setAlarm()才能调度下一次执行 - 仅在有工作时才调度 alarm — 避免让每个 Durable Object 以短间隔(秒级)唤醒,因为每次 alarm 调用都会产生费用
import { DurableObject } from "cloudflare:workers";
export class GameMatch extends DurableObject {
async startGame(durationMs = 60000) {
await this.ctx.storage.put("gameStarted", Date.now());
await this.ctx.storage.put("gameActive", true);
// Schedule the game to end after the duration
await this.ctx.storage.setAlarm(Date.now() + durationMs);
}
// Called when the alarm fires
async alarm(alarmInfo) {
const isActive = await this.ctx.storage.get("gameActive");
if (!isActive) {
return; // Game was already ended
}
// End the game
await this.ctx.storage.put("gameActive", false);
await this.ctx.storage.put("gameEnded", Date.now());
// Calculate final scores, notify players, etc.
try {
await this.calculateFinalScores();
} catch (err) {
// If we're almost out of retries but still have work to do, schedule a new alarm
// rather than letting our retries run out to ensure we keep getting invoked.
if (alarmInfo && alarmInfo.retryCount >= 5) {
await this.ctx.storage.setAlarm(Date.now() + 30 * 1000);
return;
}
throw err;
}
// Schedule the next alarm only if there's more work to do
// In this case, schedule cleanup in 24 hours
await this.ctx.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000);
}
async calculateFinalScores() {
// Game ending logic
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
GAME_MATCH: DurableObjectNamespace<GameMatch>;
}
export class GameMatch extends DurableObject<Env> {
async startGame(durationMs: number = 60000) {
await this.ctx.storage.put("gameStarted", Date.now());
await this.ctx.storage.put("gameActive", true);
// Schedule the game to end after the duration
await this.ctx.storage.setAlarm(Date.now() + durationMs);
}
// Called when the alarm fires
async alarm(alarmInfo?: AlarmInvocationInfo) {
const isActive = await this.ctx.storage.get<boolean>("gameActive");
if (!isActive) {
return; // Game was already ended
}
// End the game
await this.ctx.storage.put("gameActive", false);
await this.ctx.storage.put("gameEnded", Date.now());
// Calculate final scores, notify players, etc.
try {
await this.calculateFinalScores();
} catch (err) {
// If we're almost out of retries but still have work to do, schedule a new alarm
// rather than letting our retries run out to ensure we keep getting invoked.
if (alarmInfo && alarmInfo.retryCount >= 5) {
await this.ctx.storage.setAlarm(Date.now() + 30 * 1000);
return;
}
throw err;
}
// Schedule the next alarm only if there's more work to do
// In this case, schedule cleanup in 24 hours
await this.ctx.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000);
}
private async calculateFinalScores() {
// Game ending logic
}
}在极少数情况下,alarm 可能触发不止一次。您的 alarm() 处理器应能安全地多次运行而不引发问题。
import { DurableObject } from "cloudflare:workers";
export class Subscription extends DurableObject {
async alarm() {
// ✅ Good: Check state before performing the action
const lastRenewal = await this.ctx.storage.get("lastRenewal");
const renewalPeriod = 30 * 24 * 60 * 60 * 1000; // 30 days
// If we already renewed recently, don't do it again
if (lastRenewal && Date.now() - lastRenewal < renewalPeriod - 60000) {
console.log("Already renewed recently, skipping");
return;
}
// Perform the renewal
const success = await this.processRenewal();
if (success) {
// Record the renewal time
await this.ctx.storage.put("lastRenewal", Date.now());
// Schedule the next renewal
await this.ctx.storage.setAlarm(Date.now() + renewalPeriod);
} else {
// Retry in 1 hour
await this.ctx.storage.setAlarm(Date.now() + 60 * 60 * 1000);
}
}
async processRenewal() {
// Payment processing logic
return true;
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
SUBSCRIPTION: DurableObjectNamespace<Subscription>;
}
export class Subscription extends DurableObject<Env> {
async alarm() {
// ✅ Good: Check state before performing the action
const lastRenewal = await this.ctx.storage.get<number>("lastRenewal");
const renewalPeriod = 30 * 24 * 60 * 60 * 1000; // 30 days
// If we already renewed recently, don't do it again
if (lastRenewal && Date.now() - lastRenewal < renewalPeriod - 60000) {
console.log("Already renewed recently, skipping");
return;
}
// Perform the renewal
const success = await this.processRenewal();
if (success) {
// Record the renewal time
await this.ctx.storage.put("lastRenewal", Date.now());
// Schedule the next renewal
await this.ctx.storage.setAlarm(Date.now() + renewalPeriod);
} else {
// Retry in 1 hour
await this.ctx.storage.setAlarm(Date.now() + 60 * 60 * 1000);
}
}
private async processRenewal(): Promise<boolean> {
// Payment processing logic
return true;
}
}要完全清除 Durable Object 的存储,请调用 deleteAll()。仅删除单个键或删除表是不够的,因为某些内部元数据可能仍然保留。兼容性日期早于 2026-02-24 且已设置 alarm 的 Workers,应先用 deleteAlarm() 删除 alarm。
import { DurableObject } from "cloudflare:workers";
export class ChatRoom extends DurableObject {
async clearStorage() {
// Delete all storage, including any set alarm
await this.ctx.storage.deleteAll();
// The Durable Object instance still exists, but with empty storage
// A subsequent request will find no data
}
}import { DurableObject } from "cloudflare:workers";
export interface Env {
CHAT_ROOM: DurableObjectNamespace<ChatRoom>;
}
export class ChatRoom extends DurableObject<Env> {
async clearStorage() {
// Delete all storage, including any set alarm
await this.ctx.storage.deleteAll();
// The Durable Object instance still exists, but with empty storage
// A subsequent request will find no data
}
}Durable Object 可能因部署、不活动或运行时决策而随时关闭。不要依赖关闭钩子(系统不提供),而应设计应用以增量方式写入状态。
不提供关闭钩子或关闭前运行的生命周期回调,因为 Cloudflare 无法保证这些钩子在所有情况下都会执行,且外部软件可能过度依赖这些(不可靠的)钩子。
与其依赖关闭钩子,你可以定期写入存储,以便从关闭中优雅恢复。
例如,如果你正在处理数据流并需要保存进度,应在处理过程中写入位置,而不是等到最后才持久化:
// Good: Write progress as you go
async processData(data) {
data.forEach(async (item, index) => {
await this.processItem(item);
// Save progress frequently
await this.ctx.storage.put("lastProcessedIndex", index);
});
}虽然这可能感觉违反直觉,但 Durable Object 存储写入快速且同步,因此你可以以极小的性能顾虑持久化状态。
这种方法确保 Durable Object 可以从任何点安全恢复,即使意外关闭也是如此。
由单个 Durable Object 处理所有流量会成为瓶颈。虽然异步操作允许请求交错,但所有同步 JavaScript 执行都是单线程的,而存储操作提供的串行化保证会限制吞吐量。
一个常见错误是使用 Durable Object 做全局速率限制或全局计数器。这会将所有流量汇集到单个实例:
import { DurableObject } from "cloudflare:workers";
// 🔴 Bad: Global rate limiter - ALL requests go through one instance
export class RateLimiter extends DurableObject {
async checkLimit(ip) {
const key = `rate:${ip}`;
const count = (await this.ctx.storage.get(key)) ?? 0;
await this.ctx.storage.put(key, count + 1);
return count < 100;
}
}
// 🔴 Bad: Always using the same ID creates a global bottleneck
export default {
async fetch(request, env) {
// Every single request to your application goes through this one DO
const limiter = env.RATE_LIMITER.get(env.RATE_LIMITER.idFromName("global"));
const ip = request.headers.get("CF-Connecting-IP") ?? "unknown";
const allowed = await limiter.checkLimit(ip);
if (!allowed) {
return new Response("Rate limited", { status: 429 });
}
return new Response("OK");
},
};import { DurableObject } from "cloudflare:workers";
export interface Env {
RATE_LIMITER: DurableObjectNamespace<RateLimiter>;
}
// 🔴 Bad: Global rate limiter - ALL requests go through one instance
export class RateLimiter extends DurableObject<Env> {
async checkLimit(ip: string): Promise<boolean> {
const key = `rate:${ip}`;
const count = (await this.ctx.storage.get<number>(key)) ?? 0;
await this.ctx.storage.put(key, count + 1);
return count < 100;
}
}
// 🔴 Bad: Always using the same ID creates a global bottleneck
export default {
async fetch(request: Request, env: Env): Promise<Response> {
// Every single request to your application goes through this one DO
const limiter = env.RATE_LIMITER.get(
env.RATE_LIMITER.idFromName("global")
);
const ip = request.headers.get("CF-Connecting-IP") ?? "unknown";
const allowed = await limiter.checkLimit(ip);
if (!allowed) {
return new Response("Rate limited", { status: 429 });
}
return new Response("OK");
},
};此模式无法扩展。随着流量增长,单个 Durable Object 会成为卡点。相反,请识别应用中自然的协调边界(按用户、按房间、按文档),并为每个边界创建独立的 Durable Object。
使用 @cloudflare/vitest-pool-workers 测试 Durable Objects。该集成提供了直接访问实例的工具。
import { env } from "cloudflare:workers";
import { runInDurableObject, runDurableObjectAlarm } from "cloudflare:test";
import { describe, it, expect } from "vitest";
describe("ChatRoom", () => {
it("should send and retrieve messages", async () => {
const id = env.CHAT_ROOM.idFromName("test-room");
const stub = env.CHAT_ROOM.get(id);
// Call RPC methods directly on the stub
await stub.sendMessage("user-1", "Hello!");
await stub.sendMessage("user-2", "Hi there!");
const messages = await stub.getMessages(10);
expect(messages).toHaveLength(2);
});
it("can access instance internals and trigger alarms", async () => {
const id = env.CHAT_ROOM.idFromName("test-room");
const stub = env.CHAT_ROOM.get(id);
// Access storage directly for verification
await runInDurableObject(stub, async (instance, state) => {
const count = state.storage.sql
.exec("SELECT COUNT(*) as count FROM messages")
.one();
expect(count.count).toBe(2);
});
// Trigger alarms immediately without waiting
const alarmRan = await runDurableObjectAlarm(stub);
expect(alarmRan).toBe(false); // No alarm was scheduled
});
});import { env } from "cloudflare:workers";
import {
runInDurableObject,
runDurableObjectAlarm,
} from "cloudflare:test";
import { describe, it, expect } from "vitest";
describe("ChatRoom", () => {
it("should send and retrieve messages", async () => {
const id = env.CHAT_ROOM.idFromName("test-room");
const stub = env.CHAT_ROOM.get(id);
// Call RPC methods directly on the stub
await stub.sendMessage("user-1", "Hello!");
await stub.sendMessage("user-2", "Hi there!");
const messages = await stub.getMessages(10);
expect(messages).toHaveLength(2);
});
it("can access instance internals and trigger alarms", async () => {
const id = env.CHAT_ROOM.idFromName("test-room");
const stub = env.CHAT_ROOM.get(id);
// Access storage directly for verification
await runInDurableObject(stub, async (instance, state) => {
const count = state.storage.sql
.exec<{ count: number }>("SELECT COUNT(*) as count FROM messages")
.one();
expect(count.count).toBe(2);
});
// Trigger alarms immediately without waiting
const alarmRan = await runDurableObjectAlarm(stub);
expect(alarmRan).toBe(false); // No alarm was scheduled
});
});在 vitest.config.ts 中配置 Vitest:
import { cloudflareTest } from "@cloudflare/vitest-pool-workers";
import { defineConfig } from "vitest/config";
export default defineConfig({
plugins: [
cloudflareTest({
wrangler: { configPath: "./wrangler.jsonc" },
}),
],
});对于数据 schema 变更,请在构造函数中使用 blockConcurrencyWhile() 运行 schema 迁移。对于类重命名或删除,请更改 Wrangler 配置文件中 exports 字段的类条目:
{
"exports": {
// Rename a class — also add a live entry for the new name
"OldChatRoom": { "type": "durable-object", "state": "renamed", "renamed_to": "ChatRoom" },
"ChatRoom": { "type": "durable-object", "storage": "sqlite" },
// Delete a class (removes all data!)
"DeprecatedRoom": { "type": "durable-object", "state": "deleted" }
}
}[exports.OldChatRoom]
type = "durable-object"
state = "renamed"
renamed_to = "ChatRoom"
[exports.ChatRoom]
type = "durable-object"
storage = "sqlite"
[exports.DeprecatedRoom]
type = "durable-object"
state = "deleted"有关类生命周期变更的更多详情,请参阅 Durable Object 类导出;有关包括 SQLite 查询和 alarm 测试在内的完整测试模式,请参阅使用 Durable Objects 进行测试。
- Workers 最佳实践:适用于调用 Durable Objects 的 Workers 的请求处理、可观测性和安全代码模式。
- Workflows 设计准则:持久、多步骤 Workflows 的最佳实践——当您将 Workflows 与 Durable Objects 结合用于长时间编排时很有用。