Cloudflare KV 与 Durable Objects 深度指南：键值存储与有状态边缘计算

// wrangler.toml
[[kv_namespaces]]
binding = "MY_KV"
id = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

// src/index.ts
export interface Env {
  MY_KV: KVNamespace;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    // 写入
    await env.MY_KV.put('user:123', JSON.stringify({ name: 'Alice', age: 30 }));
    await env.MY_KV.put('session:abc', 'data', { expirationTtl: 3600 }); // 1 小时后过期

    // 读取
    const user = await env.MY_KV.get('user:123');
    const userWithMetadata = await env.MY_KV.getWithMetadata('user:123');
    // userWithMetadata.value = JSON string
    // userWithMetadata.metadata = { ...custom metadata... }

    // 删除
    await env.MY_KV.delete('user:123');

    // 列表（前缀匹配）
    const list = await env.MY_KV.list({ prefix: 'user:' });
    // list.keys = [{ name: 'user:123', expiration: ... }, ...]

    return Response.json({ user: user ? JSON.parse(user) : null });
  },
};

// 方式一：Name-based ID（推荐）
// 基于字符串生成确定性 ID，相同 name 始终路由到同一个 DO
const id = env.MY_DO.idFromName('room:game-123');
const doStub = env.MY_DO.get(id);

// 方式二：Unique ID（UUID）
// 生成全新的唯一 ID，用于创建全新对象
const id = env.MY_DO.newUniqueId();
const doStub = env.MY_DO.get(id);

// src/index.ts
export interface Env {
  MY_DO: DurableObjectNamespace;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const roomId = url.pathname.split('/')[2]; // /room/:id

    // 获取或创建 DO
    const id = env.MY_DO.idFromName(`room:${roomId}`);
    const room = env.MY_DO.get(id);

    // 转发请求到 DO
    return room.fetch(request);
  },
};

// src/Room.ts
export class ChatRoom implements DurableObject {
  private state: DurableObjectState;
  private sessions: Map<WebSocket, string> = new Map(); // 连接的 WebSocket

  constructor(state: DurableObjectState) {
    this.state = state;
  }

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    switch (url.pathname) {
      case '/connect': {
        // WebSocket 升级
        const [client, server] = Object.values(new WebSocketPair());
        this.handleSession(server);
        return new Response(null, { status: 101, webSocket: client });
      }

      case '/message': {
        // HTTP 发送消息（广播给所有连接的 WebSocket）
        const { message } = await request.json();
        this.broadcast(message);
        return Response.json({ sent: true });
      }

      case '/state': {
        // 读取持久化状态
        const stored = await this.state.storage.get('messages') || [];
        return Response.json({ messages: stored });
      }

      default:
        return new Response('Not Found', { status: 404 });
    }
  }

  private handleSession(ws: WebSocket) {
    this.state.acceptWebSocket(ws); // 接受 WebSocket
    this.sessions.set(ws, 'anonymous');

    // 发送历史消息
    this.state.storage.get('messages').then((history: any) => {
      if (history) ws.send(JSON.stringify({ type: 'history', data: history }));
    });
  }

  private async broadcast(message: string) {
    // 保存到持久化存储
    const history = (await this.state.storage.get('messages')) || [];
    history.push({ message, time: Date.now() });
    await this.state.storage.put('messages', history);

    // 广播给所有 WebSocket
    for (const [ws] of this.sessions) {
      ws.send(JSON.stringify({ type: 'message', data: message }));
    }
  }
}

# wrangler.toml
[[durable_objects.bindings]]
name = "MY_DO"
class_name = "ChatRoom"
script_name = "my-worker"

export class Counter implements DurableObject {
  private state: DurableObjectState;

  constructor(state: DurableObjectState) {
    this.state = state;
  }

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === '/schedule') {
      // 设置 30 秒后触发的 alarm
      await this.state.storage.setAlarm(Date.now() + 30000);
      return Response.json({ scheduled: true });
    }

    return Response.json({ count: await this.state.storage.get('count') || 0 });
  }

  async alarm() {
    // 定时执行的任务
    const count = (await this.state.storage.get('count') || 0) as number;
    await this.state.storage.put('count', count + 1);
    console.log(`Counter incremented to ${count + 1}`);
  }
}

export class Counter implements DurableObject {
  constructor(private state: DurableObjectState) {}

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    if (url.pathname === '/increment') {
      let count = (await this.state.storage.get('count')) || 0;
      count++;
      await this.state.storage.put('count', count);
      return Response.json({ count });
    }
    return Response.json({ count: await this.state.storage.get('count') || 0 });
  }
}

// 外部调用
const id = env.COUNTER.idFromName('global-counter');
const counter = env.COUNTER.get(id);
const res = await counter.fetch('https://fake-url/increment');

// 登录时创建会话
await env.SESSIONS.put(
  `session:${sessionId}`,
  JSON.stringify({ userId, role, loginAt: Date.now() }),
  { expirationTtl: 86400 } // 24 小时过期
);

// 验证会话
const session = await env.SESSIONS.get(`session:${sessionId}`);
if (!session) return new Response('Unauthorized', { status: 401 });
const { userId, role } = JSON.parse(session);

export class RateLimiter implements DurableObject {
  constructor(private state: DurableObjectState) {}

  async fetch(request: Request): Promise<Response> {
    const ip = request.headers.get('CF-Connecting-IP');
    const key = `rate:${ip}`;
    const window = 60 * 1000; // 1 分钟窗口
    const limit = 100; // 100 请求/分钟

    const now = Date.now();
    const data = (await this.state.storage.get(key)) || { count: 0, resetAt: now + window };

    if (now > data.resetAt) {
      data.count = 0;
      data.resetAt = now + window;
    }

    data.count++;
    await this.state.storage.put(key, data);

    if (data.count > limit) {
      return new Response('Rate Limited', { status: 429 });
    }

    return Response.json({ remaining: limit - data.count });
  }
}

export class Whiteboard implements DurableObject {
  private webSockets = new Set<WebSocket>();

  constructor(private state: DurableObjectState) {}

  async fetch(request: Request): Promise<Response> {
    const upgradeHeader = request.headers.get('Upgrade');
    if (upgradeHeader !== 'websocket') {
      return new Response('Expected websocket', { status: 400 });
    }

    const [client, server] = Object.values(new WebSocketPair());
    this.state.acceptWebSocket(server);
    this.webSockets.add(server);

    server.addEventListener('message', async (msg) => {
      const data = JSON.parse(msg.data as string);

      // 保存历史
      const history = (await this.state.storage.get('strokes')) || [];
      history.push(data);
      await this.state.storage.put('strokes', history);

      // 广播给其他客户端
      for (const ws of this.webSockets) {
        if (ws !== server && ws.readyState === WebSocket.READY_STATE_OPEN) {
          ws.send(JSON.stringify(data));
        }
      }
    });

    return new Response(null, { status: 101, webSocket: client });
  }
}