协调并发预订

简介

一个工作坊还剩 4 个座位，但可能有 10 个人几乎同时点击 Reserve。如果每个请求都先读取 reserved = 0，暂停片刻，再写入 reserved = 1，应用就会丢失本应成功的预订。另一种错误设计则可能批准超过工作坊实际容量的座位数。多个尚未完成的异步任务相互交错，这种现象称为交错执行（interleaving）。

在本实验中，每个经过验证的工作坊名称都会选择一个 Durable Object。该对象拥有容量记录，并为每次尝试保存一条持久化记录。它的预订方法会在一个同步 SQLite 事务中完成容量检查、计数器更新和尝试记录。并发调用可以同时到达，但任何调用都不会看到只完成了一半的状态转换。

你将学习三个相关的边界：

• 并发（Concurrency）表示多个操作在同一时间段内处于进行中；它不要求使用多个 JavaScript 线程。
• 原子性（Atomicity）表示其他操作看到的要么是完整的状态变化，要么完全看不到这次变化。
• 安全初始化（Safe initialization）会在缺少记录时创建记录，但不会覆盖已经包含预订数据的记录。

你将发送并发的本地和云端测试请求，将接受数与拒绝数同持久化状态进行比较，重启本地运行时，重新部署云端 Worker，检查 Dashboard，并删除所有临时资源。

直接开始本课程前，请先完成 将 LabEx 连接到你的 Cloudflare 账户。 每台全新的 VM 都需要单独完成 Wrangler 授权。你应该已经理解 O01–O02 中介绍的稳定 Durable Object 名称、RPC 以及基于 SQLite 的状态。

Cloudflare 目前在 Workers Free 计划中支持基于 SQLite 的 Durable Objects。本实验会创建一个临时类命名空间、多个很小的命名对象以及有上限的请求批次。安装过程会在 /home/labex/project/concurrent-reservations 中安装 Node.js 22.22.0 和项目本地的 Wrangler 4.132.0；它不会授权 Cloudflare、创建命名空间、部署 Worker 或创建预订。

授权 VM 并配置工作坊命名空间

在此步骤中，你将授权全新的 VM，并声明一个基于 SQLite 的 Durable Object 类。每个工作坊名称都会在此命名空间中选择不同的对象。

进入项目目录，确认固定的 Wrangler 版本，然后开始设备授权：

cd /home/labex/project/concurrent-reservations
npx wrangler --version
npx wrangler login --device --browser=false

预期 Wrangler 版本为 4.132.0。在浏览器中打开显示的 Cloudflare URL，输入短代码，确认要使用的学习账户并完成授权。只有在 Wrangler 报告成功后，才能返回实验。不要将密码或令牌粘贴到实验中。

读取安全的身份字段，选择已确认的账户 ID，但不要打印该 ID，然后生成唯一的 Worker 名称：

WHOAMI="$(npx wrangler whoami --json)"
printf '%s\n' "$WHOAMI" | jq '{loggedIn, authType, accounts: [.accounts[] | {name}]}'
ACCOUNT_ID="$(printf '%s\n' "$WHOAMI" | jq -r '.accounts[] | select(.name == "LabEx Learning") | .id')"
test -n "$ACCOUNT_ID"
RUN="labex-c10-o03-$(openssl rand -hex 6)"
printf '%s\n' "$RUN"

如果你的专用学习账户使用了其他显示名称，请替换为你确认过的名称。创建配置文件：

cat > wrangler.jsonc <<JSON
{
  "\$schema": "./node_modules/wrangler/config-schema.json",
  "name": "$RUN",
  "account_id": "$ACCOUNT_ID",
  "main": "src/index.js",
  "compatibility_date": "2026-09-18",
  "workers_dev": true,
  "preview_urls": false,
  "observability": {
    "enabled": true,
    "head_sampling_rate": 1
  },
  "durable_objects": {
    "bindings": [
      { "name": "WORKSHOPS", "class_name": "WorkshopReservations" }
    ]
  },
  "exports": {
    "WorkshopReservations": { "type": "durable-object", "storage": "sqlite" }
  }
}
JSON

WORKSHOPS 是 Worker 的入口命名空间绑定。类导出会为每个命名工作坊提供一个独立的 SQLite 数据库。在部署之前，云端不会创建任何资源。

实现原子预订状态转换

在此步骤中，你将创建持久化的容量表和尝试记录表，然后实现一次原子预订状态转换。

每当 Cloudflare 创建或重启内存中的类实例时，都会运行构造函数。CREATE TABLE IF NOT EXISTS 会安全地重新创建缺失的表结构。INSERT ... ON CONFLICT DO NOTHING 只会在四座容量记录不存在时插入它；如果 reserved 已经有值，则不会将其重置为零。

创建 Worker 入口文件：

cat > src/index.js <<'JS'
import { DurableObject } from "cloudflare:workers";

export class WorkshopReservations extends DurableObject {
  constructor(ctx, env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      this.ctx.storage.sql.exec(`
        CREATE TABLE IF NOT EXISTS workshop_state (
          singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
          capacity INTEGER NOT NULL CHECK (capacity > 0),
          reserved INTEGER NOT NULL CHECK (reserved >= 0 AND reserved <= capacity)
        )
      `);
      this.ctx.storage.sql.exec(`
        CREATE TABLE IF NOT EXISTS reservation_attempts (
          request_id TEXT PRIMARY KEY,
          seats INTEGER NOT NULL CHECK (seats > 0),
          status TEXT NOT NULL CHECK (status IN ('accepted', 'rejected')),
          reserved_after INTEGER NOT NULL,
          created_at INTEGER NOT NULL
        )
      `);
      this.ctx.storage.sql.exec(`
        INSERT INTO workshop_state (singleton, capacity, reserved)
        VALUES (1, 4, 0)
        ON CONFLICT(singleton) DO NOTHING
      `);
    });
  }

  reserve(requestId, seats) {
    return this.ctx.storage.transactionSync(() => {
      const previous = this.ctx.storage.sql.exec(
        `SELECT request_id AS requestId, seats, status, reserved_after AS reserved
         FROM reservation_attempts WHERE request_id = ?`,
        requestId
      ).toArray()[0];
      if (previous) {
        const state = this.ctx.storage.sql.exec(
          `SELECT capacity FROM workshop_state WHERE singleton = 1`
        ).one();
        return { ...previous, capacity: state.capacity, replayed: true };
      }

      const updated = this.ctx.storage.sql.exec(
        `UPDATE workshop_state
         SET reserved = reserved + ?
         WHERE singleton = 1 AND reserved + ? <= capacity
         RETURNING capacity, reserved`,
        seats,
        seats
      ).toArray();
      const accepted = updated.length === 1;
      const state = accepted ? updated[0] : this.ctx.storage.sql.exec(
        `SELECT capacity, reserved FROM workshop_state WHERE singleton = 1`
      ).one();
      const status = accepted ? "accepted" : "rejected";

      this.ctx.storage.sql.exec(
        `INSERT INTO reservation_attempts
         (request_id, seats, status, reserved_after, created_at)
         VALUES (?, ?, ?, ?, ?)`,
        requestId,
        seats,
        status,
        state.reserved,
        Date.now()
      );
      return { requestId, seats, status, capacity: state.capacity, reserved: state.reserved, replayed: false };
    });
  }

  getStatus() {
    return this.ctx.storage.sql.exec(`
      SELECT
        s.capacity,
        s.reserved,
        COUNT(CASE WHEN a.status = 'accepted' THEN 1 END) AS acceptedRequests,
        COUNT(CASE WHEN a.status = 'rejected' THEN 1 END) AS rejectedRequests,
        COALESCE(SUM(CASE WHEN a.status = 'accepted' THEN a.seats ELSE 0 END), 0) AS acceptedSeats
      FROM workshop_state AS s
      LEFT JOIN reservation_attempts AS a ON 1 = 1
      WHERE s.singleton = 1
      GROUP BY s.capacity, s.reserved
    `).one();
  }
}

function json(data, status = 200) {
  return Response.json(data, { status });
}

function workshopRoute(pathname) {
  const match = pathname.match(/^\/workshops\/([^/]+)\/(reservations|status)$/);
  if (!match) return { error: "not_found", status: 404 };
  let workshop;
  try {
    workshop = decodeURIComponent(match[1]);
  } catch {
    return { error: "invalid_workshop_name", status: 400 };
  }
  if (!/^[a-z][a-z0-9-]{0,31}$/.test(workshop)) {
    return { error: "invalid_workshop_name", status: 400 };
  }
  return { workshop, action: match[2] };
}

function validReservation(value) {
  return value &&
    /^[a-z][a-z0-9-]{2,47}$/.test(value.requestId) &&
    Number.isInteger(value.seats) &&
    value.seats >= 1 && value.seats <= 4;
}

export default {
  async fetch(request, env) {
    const url = new URL(request.url);
    if (request.method === "GET" && url.pathname === "/health") {
      return json({ status: "ok" });
    }

    const parsed = workshopRoute(url.pathname);
    if (parsed.error) return json({ error: parsed.error }, parsed.status);

    if (request.method === "GET" && parsed.action === "status") {
      const stub = env.WORKSHOPS.getByName(parsed.workshop);
      const state = await stub.getStatus();
      return json({ workshop: parsed.workshop, ...state });
    }
    if (request.method === "POST" && parsed.action === "reservations") {
      let body;
      try {
        body = await request.json();
      } catch {
        return json({ error: "invalid_json" }, 400);
      }
      if (!validReservation(body)) return json({ error: "invalid_reservation" }, 400);
      const stub = env.WORKSHOPS.getByName(parsed.workshop);
      const result = await stub.reserve(body.requestId, body.seats);
      console.log(JSON.stringify({ event: "reservation_decided", workshop: parsed.workshop, requestId: body.requestId, status: result.status, reserved: result.reserved }));
      return json({ workshop: parsed.workshop, ...result }, result.status === "accepted" ? 201 : 409);
    }
    return json({ error: "method_not_allowed" }, 405);
  }
};
JS

transactionSync() 只接受同步的存储操作。带条件的 UPDATE 只有在请求的座位数仍未超过容量时才会更新计数器，RETURNING 会读取同一条语句产生的值。尝试记录也会在同一个事务中提交。重复的 requestId 会返回第一次的决定，而不会再次消耗容量。

运行确定性的路由测试和实际打包检查：

NODE_NO_WARNINGS=1 node --experimental-loader ./test/cloudflare-loader.mjs --test test/worker.test.mjs
npx wrangler deploy --dry-run

预期两个测试通过，并且 dry run 成功。不会创建远程资源。

并发发送 10 个本地请求

在此步骤中，10 个预订命令会同时为同一个四座工作坊处理。xargs -P 10 会同时启动最多 10 个 Shell 进程；这些进程完成的顺序没有固定要求。

使用明确的持久化目录启动本地运行时：

npx wrangler dev --port 8787 --persist-to .labex/local-state > .labex/dev.log 2>&1 &
echo $! > .labex/dev.pid
for attempt in $(seq 1 30); do
  curl --silent --fail http://127.0.0.1:8787/health && break
  sleep 1
done

向同一个稳定对象发送 10 个单座位尝试。每个进程都会写入独立的响应文件，因此并发终端输出不会混在一起：

rm -f .labex/local-response-*.json
seq 1 10 | xargs -P 10 -I{} sh -c '
  curl --silent \
    --request POST http://127.0.0.1:8787/workshops/launch-day/reservations \
    --header "content-type: application/json" \
    --data "{\"requestId\":\"request-$1\",\"seats\":1}" \
    > ".labex/local-response-$1.json"
' _ {}

对于被拒绝的预订，HTTP 409 是预期的应用响应。curl --silent 仍会保存 JSON 响应正文，因此即使工作坊已满，也不会把它当成传输失败。将所有决定合并为一个数组进行检查：

jq -s 'sort_by(.requestId)' .labex/local-response-*.json
jq -s '{
  accepted: map(select(.status == "accepted")) | length,
  rejected: map(select(.status == "rejected")) | length,
  highestReserved: map(.reserved) | max
}' .labex/local-response-*.json

具体哪些请求 ID 被接受可能不同，因为请求到达顺序没有保证。但不变量不会改变：恰好 4 个请求被接受，6 个请求被拒绝，并且没有任何响应报告已预订座位数超过 4。

从对象中读取持久化的总数：

curl --silent http://127.0.0.1:8787/workshops/launch-day/status | jq

预期容量为 4，已预订数为 4，接受的请求数为 4，拒绝的请求数为 6，接受的座位数为 4。响应文件说明每个请求的结果；状态记录则证明这些总数与持久化状态一致。

重启而不重置容量

在此步骤中，你将通过停止 Wrangler 移除内存中的类实例，然后使用同一个数据库启动新的运行时，证明初始化不会恢复容量。

停止并重新启动进程：

kill "$(cat .labex/dev.pid)"
wait "$(cat .labex/dev.pid)" 2>/dev/null || true
npx wrangler dev --port 8787 --persist-to .labex/local-state > .labex/dev-restarted.log 2>&1 &
echo $! > .labex/dev.pid
for attempt in $(seq 1 30); do
  curl --silent --fail http://127.0.0.1:8787/health && break
  sleep 1
done

在做出下一次决定前读取工作坊状态：

curl --silent http://127.0.0.1:8787/workshops/launch-day/status | jq

结果必须仍然报告 reserved: 4。构造函数再次运行了，但 ON CONFLICT DO NOTHING 保留了已有记录。

重放第一个请求 ID，然后在工作坊已满时提交一个新请求：

curl --silent --request POST http://127.0.0.1:8787/workshops/launch-day/reservations \
  --header 'content-type: application/json' \
  --data '{"requestId":"request-1","seats":1}' | jq
curl --silent --request POST http://127.0.0.1:8787/workshops/launch-day/reservations \
  --header 'content-type: application/json' \
  --data '{"requestId":"request-after-restart","seats":1}' | jq
curl --silent http://127.0.0.1:8787/workshops/launch-day/status | jq

重放结果中应包含 replayed: true，并且不会新增尝试记录。新 ID 会再次被拒绝。最终总数仍为 4 个已接受座位，拒绝请求数增加到 7 个。

体验云端并发

在此步骤中，你将停止本地运行时，部署命名空间，然后针对 Cloudflare 重复有上限的并发测试。

停止本地进程并进行部署：

kill "$(cat .labex/dev.pid)"
wait "$(cat .labex/dev.pid)" 2>/dev/null || true
DEPLOY_OUTPUT="$(npx wrangler deploy 2>&1 | tee /dev/tty)"
APP_URL="$(printf '%s\n' "$DEPLOY_OUTPUT" | grep -Eo 'https://[a-z0-9.-]+\.workers\.dev' | tail -1)"
test -n "$APP_URL"
printf '%s\n' "$APP_URL"

Worker 路由及其刚刚协调完成的 Durable Object 命名空间可能不会在同一时刻可用。先轮询一个对象的实际读取结果，确认其 JSON 符合预期，然后等待经过测试的短暂稳定时间，再创建另一个对象：

for attempt in $(seq 1 30); do
  if curl --silent --fail "$APP_URL/workshops/readiness/status" |
    jq -e '.capacity == 4 and .reserved == 0' >/dev/null; then
    break
  fi
  sleep 1
done
curl --silent --fail "$APP_URL/workshops/readiness/status" |
  jq -e '.capacity == 4 and .reserved == 0'
sleep 5

向 cloud-launch 发送 10 个并发请求：

rm -f .labex/cloud-response-*.json
seq 1 10 | xargs -P 10 -I{} sh -c '
  curl --silent \
    --request POST "$0/workshops/cloud-launch/reservations" \
    --header "content-type: application/json" \
    --data "{\"requestId\":\"cloud-request-$1\",\"seats\":1}" \
    > ".labex/cloud-response-$1.json"
' "$APP_URL" {}

比较响应总数和持久化状态：

jq -s '{
  accepted: map(select(.status == "accepted")) | length,
  rejected: map(select(.status == "rejected")) | length,
  highestReserved: map(.reserved) | max
}' .labex/cloud-response-*.json
curl --silent "$APP_URL/workshops/cloud-launch/status" | jq

云端不变量应与本地结果一致：4 个接受，6 个拒绝，reserved: 4。运行独立检查。该检查会检查绑定和命名空间，验证 cloud-launch，然后向另一个具有本次运行唯一名称的工作坊发送 12 个并发请求：

python3 .labex/verify.py deployed

重新部署并检查预订协调

在此步骤中，你将重新部署未修改的 Worker。这可能会替换内存中的类实例，因此构造函数可能再次运行。持久化的容量记录必须仍然显示已满。

重新部署，然后通过新请求读取 cloud-launch：

npx wrangler deploy
curl --silent "$APP_URL/workshops/cloud-launch/status" | jq

预期容量仍为 4，已预订数仍为 4，接受请求数为 4，拒绝请求数为 6。这次云端重启检查与本地重启得出相同结论：安全初始化会创建缺失的状态，但绝不会覆盖已经建立的状态。

打开 Cloudflare Dashboard 并选择同一个账户。进入 Workers & Pages，打开名称完全匹配的 labex-c10-o03-... Worker，然后选择 Bindings。确认 WORKSHOPS 指向经过测试的 WorkshopReservations 命名空间。

截图中显示的后缀属于通过验证的编写运行。你生成的后缀会不同；必须匹配的是绑定名称、类型和目标类。

打开该命名空间并选择 Overview。Storage: SQL 表示用于保存容量表和尝试记录表的后端。

现在打开 Logs。成功的 WorkshopReservations.jsrpc 行表示并发批次和验证脚本发起的对象方法调用。你会看到多个对象 ID，因为 readiness、cloud-launch 以及验证脚本使用的唯一工作坊被有意隔离。日志显示调用和错误；HTTP 总数才是证明容量得到遵守的权威依据。

重新部署后，再次运行独立的云端检查：

python3 .labex/verify.py deployed

删除预订命名空间并退出登录

在此步骤中，你将永久删除临时命名空间及其工作坊数据库，删除剩余的 Worker，并撤销此 VM 的授权。

确认 $RUN 以 labex-c10-o03- 开头。创建无状态的清理入口文件：

cat > src/cleanup.js <<'JS'
export default {
  fetch() {
    return Response.json({ status: "cleanup" }, { status: 410 });
  }
};
JS

为完全相同的 Worker 和账户创建清理配置。state: "deleted" 墓碑配置只会永久删除 WorkshopReservations 类命名空间：

ACCOUNT_ID="$(node -e 'console.log(JSON.parse(require("fs").readFileSync("wrangler.jsonc", "utf8")).account_id)')"
cat > wrangler.cleanup.jsonc <<JSON
{
  "\$schema": "./node_modules/wrangler/config-schema.json",
  "name": "$RUN",
  "account_id": "$ACCOUNT_ID",
  "main": "src/cleanup.js",
  "compatibility_date": "2026-09-18",
  "workers_dev": true,
  "preview_urls": false,
  "exports": {
    "WorkshopReservations": { "type": "durable-object", "state": "deleted" }
  }
}
JSON
npx wrangler deploy --config wrangler.cleanup.jsonc

协调输出应报告 Deleted: WorkshopReservations。删除剩余的无状态 Worker，并确认只删除确切的生成名称：

npx wrangler delete --config wrangler.cleanup.jsonc

退出登录前，证明已通过身份验证确认资源不存在：

python3 .labex/verify.py deleted

只有在看到 PASS: deleted 后，才能撤销 VM 授权并检查结构化状态：

npx wrangler logout
npx wrangler whoami --json

最终 JSON 必须包含 "loggedIn": false。网络错误或身份验证错误不能作为清理完成的证据。

总结

你构建了一个有容量上限的预订服务，其中每个工作坊名称都会选择一个 Durable Object。同步 SQLite 事务将容量检查、计数器变化和尝试记录组合为一个不可分割的状态转换。10 个并发调用可以按任意顺序完成，但最终恰好接受 4 个座位，并且没有任何响应超过容量上限。

你还使用 ON CONFLICT DO NOTHING 实现了安全初始化，在不重复预订的情况下重放了一个稳定的请求 ID，并证明本地重启和云端重新部署后，持久化总数仍然一致。最后，你在 Dashboard 中将运行时证据与绑定、SQL 命名空间和 RPC 日志对应起来，然后删除确切的命名空间和 Worker，最后退出登录。