《林一的 Cloudflare 通关记》第 9 篇-终于上"正餐"了——D1 数据库搭建

《林一的 Cloudflare 通关记》第 9 篇

上周林一用 KV 把星火AI 的配置缓存搞定了，首屏加载从 3 秒降到了 800 毫秒，小王在群里连发了三个点赞表情。

但好景不长。周一早上，产品经理小王甩来一张需求文档：

故事引入

"林一，用户量涨到 500 多了，需要加几个功能：第一，用户注册登录——总不能一直靠 URL 参数传 uid 吧？第二，对话记录要能按时间分页查询；第三，首页要展示'本月生成图片最多的前 10 名用户'。"

林一看了看需求，脑子开始转：用户注册要存用户名、密码哈希、邮箱；对话记录要按时间倒序分页；排行榜要聚合统计——这些全是关系型查询，KV 的 get(key) 根本搞不定。

"张哥，看来得上真正的数据库了。"林一走到老张工位旁，"我之前实习的公司用的是阿里云 RDS，最低配也要好几百一个月。咱们这个项目零预算，总不能让我自己搭个 MySQL 吧？"

老张推了推眼镜："Cloudflare 有 D1——免费的 Serverless SQL 数据库。"

"D1？"

"对，基于 SQLite 的。"

林一脸一皱："SQLite？那个嵌入式数据库？单机文件型的那种？大学课上老师说它只能做开发测试，撑不住生产环境的并发啊。"

老张笑了："你知道 SQLite 是全球部署量最大的数据库吗？每台手机、每个浏览器里都跑着 SQLite。而且 D1 的架构和你想的那种嵌入式 SQLite 完全不一样——它是个分布式数据库。"

"分布式 SQLite？这不是矛盾吗？"

"一点都不矛盾。来，我给你好好讲讲。"

---

技术讲解

SQLite：被误解的"小"数据库

先简单过一下 SQLite 是什么。如果你已经了解，可以跳过这段。

SQLite 是一个嵌入式关系型数据库——不需要独立的服务进程，整个数据库就是一个文件。没有客户端/服务端架构，没有网络通信开销，应用程序直接通过函数调用读写数据库文件。这点和 MySQL、PostgreSQL 完全不同，后两者都需要一个独立运行的数据库服务进程。

用一句话概括：MySQL 是一个服务，SQLite 是一个库。

SQLite 的优点很明确：轻量、零配置、零运维、读取速度极快。缺点也同样明显：单机单文件、不支持网络访问、写入并发受限（写锁是数据库级别的）。

"所以林一的质疑其实有道理——如果是传统 SQLite，确实撑不住生产并发。"老张说，"但 D1 不是传统 SQLite。"

为什么 SQLite 适合边缘计算

"要理解 D1，先得想明白一个问题：为什么 Cloudflare 不用 MySQL 或 PostgreSQL，偏偏选了 SQLite？"

老张在白板上画了个对比表：

特性	MySQL/PostgreSQL	SQLite
运行方式	独立服务进程，常驻内存	嵌入式，无独立进程
网络开销	每次查询走 TCP	零网络开销，函数调用
启动时间	秒级	毫秒级
资源占用	多，需要预分配内存	极少，按需分配
部署成本	每个实例需要独立服务器	一个文件就是数据库

"Cloudflare 的 Workers 跑在全球 300 多个边缘节点上，"老张指着表说，"每个节点可能同时运行成千上万个 Worker。如果用 MySQL，每个节点都要起一个 MySQL 实例——光内存占用就受不了。而 SQLite 不需要常驻进程，Worker 需要查数据时加载，查完释放，完美契合边缘计算'按需使用'的模式。"

"简单说：MySQL 像在楼下开了一家餐厅，你要吃饭得走过去点单等菜；SQLite 像你口袋里揣了个便当盒，打开就能吃。在边缘节点上，'便当盒'才是务实的选择。"

"国内的腾讯云 CloudBase 和阿里云 Serverless 数据库走的是另一条路——用 MySQL/PostgreSQL 加连接池代理。各有取舍，D1 的方案更轻、更贴近边缘。"

Cloudflare D1 架构：分布式 SQLite

"好了，重点来了。"老张在白板上画了一张架构图：

                        ┌─────────────────────┐
                        │   Cloudflare 全球网络  │
                        │                     │
   用户请求 ──────────► │  边缘节点 (WEUR)     │
                        │  ┌───────────────┐  │
                        │  │ Worker        │  │
                        │  │   │           │  │
                        │  │   ▼           │  │
                        │  │ D1 读副本     │  │
                        │  │ (SQLite文件)  │  │
                        │  └───────┬───────┘  │
                        │          │ 异步复制   │
                        │          ▼          │
                        │  ┌───────────────┐  │
                        │  │ D1 主数据库    │  │
                        │  │ (唯一写入口)   │  │
                        │  │ (SQLite文件)  │  │
                        │  └───────────────┘  │
                        │          │          │
                        │          │ 异步复制   │
                        │          ▼          │
                        │  ┌───────────────┐  │
                        │  │ D1 读副本     │  │
                        │  │ (APAC区域)    │  │
                        │  └───────────────┘  │
                        └─────────────────────┘

"D1 的核心架构是：一个写主库 (Primary) + 多个全球读副本 (Read Replicas)。"

主数据库实例 (Primary)

主库是数据库的"原始副本"，位于全球单一位置。所有写操作——INSERT、UPDATE、DELETE——都必须路由到主库执行。主库是唯一的写入入口，保证写入的顺序性和一致性。

"就像公司只有一个财务盖章——不管哪个部门要报销，都得把单子送到财务部盖完章才算数。"

只读副本 (Read Replicas)

读副本是主库的异步复制副本，分布在 Cloudflare 全球网络的多个区域。目前支持的区域包括：

区域代码	区域
ENAM	北美东部
WNAM	北美西部
WEUR	西欧
EEUR	东欧
APAC	亚太
OC	大洋洲

读副本只能处理读查询（SELECT）。当用户发起读请求时，D1 会将请求路由到最近的副本，缩短物理距离，降低延迟。

"读副本就像公司在各地开了档案室——查阅资料不用大老远跑总部，就近去就行。但修改档案？还得回总部改，然后把复印件分发到各地。"

异步复制与副本延迟

"关键词是异步。"老张在白板上重重画了下划线，"主库写入成功后，异步把变更同步到各读副本。这意味着副本可能存在数据滞后——主库刚写入的数据，副本上可能还没出现。"

"这不是问题吗？"林一问。

"对某些场景确实是。所以 D1 提供了 Sessions API 来解决一致性问题——这个我们后面实操时讲。先记住一个结论：D1 提供顺序一致性，不是强一致性，但比最终一致性更强。"

每个副本背后是 Durable Objects

"D1 的每个数据库实例——主库和每个读副本——底层都是一个 Durable Object。"老张补充道，"Durable Objects 是 Cloudflare 的单点状态管理服务，保证了每个 D1 数据库实例是单线程顺序处理的。这听起来像限制，但实际上：如果你的查询平均耗时 1 毫秒，单实例每秒能处理约 1000 次查询。加上读副本分担读负载，吞吐量还可以继续扩展。"

"如果请求量太大，队列满了怎么办？"

"会返回 'overloaded' 错误。这时候要么优化查询、加索引让查询更快，要么水平拆分——把不同用户的数据分到不同 D1 数据库里。D1 支持每个账户创建最多 50,000 个数据库（付费计划），天生就是为水平扩展设计的。"

---

实操指导

第一步：创建 D1 数据库

"理论讲完，上手操作。"老张打开了终端。

# 创建一个名为 spark-db 的 D1 数据库
npx wrangler@latest d1 create spark-db

执行后输出类似：

✅ Successfully created DB 'spark-db'

[[d1_databases]]
binding = "SPARK_DB"
database_name = "spark-db"
database_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

"看到那个提示了吗？Wrangler 会自动把绑定配置加到你的 wrangler.jsonc（或 wrangler.toml）里。如果你选了 Yes，它已经帮你写好了。"

第二步：绑定到 Worker

如果 Wrangler 没有自动添加，手动在 wrangler.jsonc 中配置：

{
  "name": "spark-ai",
  "main": "src/index.js",
  "compatibility_date": "2024-09-25",
  "d1_databases": [
    {
      "binding": "SPARK_DB",
      "database_name": "spark-db",
      "database_id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    }
  ]
}

如果用的是 wrangler.toml：

[[d1_databases]]
binding = "SPARK_DB"
database_name = "spark-db"
database_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

"binding 的值 SPARK_DB 就是你在 Worker 代码里访问数据库的变量名。在 Worker 中通过 env.SPARK_DB 来使用它。"

第三步：用 wrangler 操作 D1

"创建好数据库后，我们先用命令行操作它，熟悉一下。"老张敲了第一条命令。

执行 SQL 文件

先创建一个 schema.sql 文件，定义星火AI 的数据库表结构：

-- schema.sql

-- 用户表
CREATE TABLE IF NOT EXISTS users (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  username TEXT UNIQUE NOT NULL,
  email TEXT UNIQUE NOT NULL,
  password_hash TEXT NOT NULL,
  created_at TEXT DEFAULT (datetime('now')),
  last_login_at TEXT
);

-- 对话记录表
CREATE TABLE IF NOT EXISTS conversations (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  user_id INTEGER NOT NULL,
  role TEXT NOT NULL CHECK(role IN ('user', 'assistant')),
  content TEXT NOT NULL,
  model TEXT,
  created_at TEXT DEFAULT (datetime('now')),
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- 图片生成记录表
CREATE TABLE IF NOT EXISTS image_generations (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  user_id INTEGER NOT NULL,
  prompt TEXT NOT NULL,
  image_url TEXT NOT NULL,
  model TEXT NOT NULL,
  created_at TEXT DEFAULT (datetime('now')),
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- 索引：加速常用查询
CREATE INDEX IF NOT EXISTS idx_conversations_user_time
  ON conversations(user_id, created_at DESC);

CREATE INDEX IF NOT EXISTS idx_image_generations_user_time
  ON image_generations(user_id, created_at DESC);

用 --local 在本地执行（本地开发用）：

npx wrangler d1 execute spark-db --local --file=./schema.sql

用 --remote 在线上数据库执行：

npx wrangler d1 execute spark-db --remote --file=./schema.sql

"--local 和 --remote 是两个关键标志，一定要分清楚。本地开发时用 --local，数据存在你电脑上的本地 SQLite 文件里；部署前用 --remote 在线上数据库执行。搞混了的话——要么本地改了线上没生效，要么线上数据被你手滑改了。"

执行单条 SQL 命令

# 查看所有表
npx wrangler d1 execute spark-db --local --command="SELECT name FROM sqlite_master WHERE type='table'"

# 插入一条测试数据
npx wrangler d1 execute spark-db --local --command="INSERT INTO users (username, email, password_hash) VALUES ('linyi', 'linyi@spark.ai', 'hash_placeholder')"

# 查询数据
npx wrangler d1 execute spark-db --local --command="SELECT * FROM users"

数据库迁移管理

"项目复杂了以后，表结构会不断变化——加字段、改索引、建新表。每次手动写 SQL 太乱，需要迁移文件管理。"

老张建了个 migrations 目录：

spark-ai/
├── migrations/
│   ├── 001_initial_schema.sql      # 初始建表
│   ├── 002_add_user_avatar.sql     # 加头像字段
│   └── 003_add_conversation_tags.sql # 对话加标签
├── schema.sql                       # 完整 schema（给新环境用）
├── wrangler.jsonc
└── src/
    └── index.js

"每个迁移文件按编号命名，记录每次结构变更。应用迁移时按顺序执行：

# 执行最新的迁移文件
npx wrangler d1 execute spark-db --remote --file=./migrations/002_add_user_avatar.sql

"D1 目前没有像 Prisma Migrate 那样的自动迁移工具，需要手动管理。但好处是简单透明——你完全知道每条 SQL 做了什么。"

第四步：在 Workers 中操作 D1

"命令行操作只是热身。真正干活的是在 Worker 代码里通过 D1 Binding API 操作数据库。"

老张打开 src/index.js，开始写代码。

核心 API：prepare / bind / run / all / first

D1 的查询模式是经典的预编译语句（Prepared Statement）三步走：

prepare(SQL) → bind(参数) → run()/all()/first()

export default {
  async fetch(request, env) {
    const url = new URL(request.url);

    // ========== 1. 查询所有用户（all 返回所有匹配行） ==========
    if (url.pathname === "/api/users" && request.method === "GET") {
      const { results } = await env.SPARK_DB
        .prepare("SELECT id, username, email, created_at FROM users ORDER BY created_at DESC LIMIT 50")
        .all();

      return Response.json({ users: results });
    }

    // ========== 2. 查询单个用户（first 返回第一行） ==========
    if (url.pathname.startsWith("/api/users/") && request.method === "GET") {
      const userId = url.pathname.split("/")[3];

      const user = await env.SPARK_DB
        .prepare("SELECT id, username, email, created_at FROM users WHERE id = ?")
        .bind(userId)
        .first();
      // first() 返回单行对象或 null

      if (!user) {
        return Response.json({ error: "用户不存在" }, { status: 404 });
      }

      return Response.json({ user });
    }

    // ========== 3. 注册新用户（run 执行写操作） ==========
    if (url.pathname === "/api/users/register" && request.method === "POST") {
      const { username, email, password_hash } = await request.json();

      // run() 执行 INSERT/UPDATE/DELETE，返回 meta 信息
      const result = await env.SPARK_DB
        .prepare("INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)")
        .bind(username, email, password_hash)
        .run();

      // result.meta.last_row_id 是新插入行的主键 ID
      return Response.json({
        success: true,
        userId: result.meta.last_row_id,
      });
    }

    return Response.json({ error: "Not Found" }, { status: 404 });
  },
};

"三步走的逻辑很清晰：prepare() 写好带 ? 占位符的 SQL——永远不要用字符串拼接拼 SQL，那是 SQL 注入的温床；bind() 把参数安全地绑上去；最后选一个执行方法。"

几个方法区别如下：

方法	返回值	适用场景
all()	{ results: [...], meta: {...} }	查询多行
first()	单行对象或 null	查询单行
run()	{ meta: {...} }	INSERT/UPDATE/DELETE，不关心返回行
raw()	数组的数组（无字段名）	高性能批量读取，省去序列化开销

批量操作：batch()

"如果一次要执行多条 SQL，用 batch() 比循环调用高效得多——它在一个事务里执行，要么全成功，要么全回滚。"

// 批量插入对话记录
if (url.pathname === "/api/conversations/batch" && request.method === "POST") {
  const { userId, messages } = await request.json();

  const stmts = messages.map(msg =>
    env.SPARK_DB
      .prepare("INSERT INTO conversations (user_id, role, content, model) VALUES (?, ?, ?, ?)")
      .bind(userId, msg.role, msg.content, msg.model)
  );

  // batch() 在单个事务中执行所有语句
  const results = await env.SPARK_DB.batch(stmts);

  return Response.json({
    success: true,
    inserted: results.length,
  });
}

对话记录分页查询

"来做一个实际需求——按时间倒序分页查询某个用户的对话记录。"

// GET /api/conversations?uid=1&page=1&size=20
if (url.pathname === "/api/conversations" && request.method === "GET") {
  const userId = url.searchParams.get("uid");
  const page = parseInt(url.searchParams.get("page") || "1");
  const size = Math.min(parseInt(url.searchParams.get("size") || "20"), 100); // 最大100条

  const offset = (page - 1) * size;

  // 查询当前页数据
  const { results } = await env.SPARK_DB
    .prepare("SELECT * FROM conversations WHERE user_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?")
    .bind(userId, size, offset)
    .all();

  // 查询总数（用于计算总页数）
  const countResult = await env.SPARK_DB
    .prepare("SELECT COUNT(*) as total FROM conversations WHERE user_id = ?")
    .bind(userId)
    .first();

  const total = countResult.total;
  const totalPages = Math.ceil(total / size);

  return Response.json({
    conversations: results,
    pagination: {
      page,
      size,
      total,
      totalPages,
    },
  });
}

"注意 LIMIT ? OFFSET ? 的参数也要用 bind() 传——SQLite 的占位符只支持值，不支持表名或 LIMIT 关键字直接拼接。"

月度排行榜（聚合查询）

"小王要的'本月生成图片最多的前 10 名用户'——一条 SQL 搞定："

// GET /api/leaderboard/monthly
if (url.pathname === "/api/leaderboard/monthly" && request.method === "GET") {
  const { results } = await env.SPARK_DB
    .prepare(`
      SELECT
        u.id,
        u.username,
        COUNT(g.id) as generation_count
      FROM users u
      JOIN image_generations g ON u.id = g.user_id
      WHERE g.created_at >= datetime('now', 'start of month')
      GROUP BY u.id
      ORDER BY generation_count DESC
      LIMIT 10
    `)
    .all();

  return Response.json({ leaderboard: results });
}

"如果这种聚合查询被频繁调用，建议把结果缓存到 KV 里——每分钟刷新一次，不必每次都查数据库。KV + D1 配合使用，才是最佳实践。"

第五步：启用读副本（Sessions API）

"前面讲了 D1 的读写分离架构，但默认情况下——即使你开了读副本——读查询也只会走主库。要真正利用读副本，必须用 Sessions API。"

export default {
  async fetch(request, env) {
    const url = new URL(request.url);

    // 创建会话——第一个查询路由到主库，确保读到最新数据
    const session = env.SPARK_DB.withSession("first-primary");

    if (url.pathname === "/api/users/profile" && request.method === "GET") {
      // 第一个查询走主库（确保拿到最新数据）
      const user = await session
        .prepare("SELECT * FROM users WHERE id = ?")
        .bind(1)
        .first();

      // 后续查询走最近的读副本（延迟更低）
      const { results } = await session
        .prepare("SELECT * FROM conversations WHERE user_id = ? ORDER BY created_at DESC LIMIT 10")
        .bind(1)
        .all();

      return Response.json({ user, recentConversations: results });
    }

    return Response.json({ error: "Not Found" }, { status: 404 });
  },
};

Sessions API 的三种启动方式：

方式	代码	第一个查询路由到	适用场景
无约束	withSession()	任意实例	不需要最新数据
从主库启动	withSession("first-primary")	主库	需要读到自己刚写入的数据
从指定版本启动	withSession(bookmark)	满足版本要求的副本	跨请求保持读取上下文

"Sessions API 的核心机制是 bookmark（书签）。每次写入后，D1 返回一个 bookmark 标识当前数据库版本；后续读查询携带这个 bookmark，副本会等到自身更新到该版本后才返回数据。这保证了'写入后立刻能读到'的一致性——这叫读己之写（Read My Own Writes）。"

"不过有个前提：你得先在 Dashboard 或通过 API 启用 Read Replication 才行。免费计划也支持读副本，不额外收费。"

D1 的限制与注意事项

"用 D1 之前，必须了解它的边界。"老张在白板上列了一张表：

限制项	免费计划	付费计划
数据库数量	10 个	50,000 个
单库最大大小	500 MB	10 GB
账户总存储	5 GB	1 TB
Time Travel 恢复	7 天	30 天
单次 Worker 调用查询数	50	1,000

"除了这些硬限制，还有几个需要特别注意的点："

1. 写入延迟：主从同步

"所有写操作都走主库，主库再异步同步到读副本。这意味着：写入操作不是'全球即时可见'的。如果你用 withSession("first-primary")，能保证自己读到自己的写入；但如果另一个用户在另一个区域读，可能要等几秒才能看到最新数据。"

"所以 D1 不适合需要跨用户强一致的场景——比如秒杀库存扣减。但用户注册、对话记录、图片元数据这些场景，完全没问题。"

2. 单库并发：单线程处理

"每个 D1 数据库实例是单线程的——查询一条一条排队处理。如果平均查询耗时 1 毫秒，每秒约 1,000 次查询；如果查询耗时 100 毫秒，每秒只有 10 次。请求量超过处理能力时，先排队，队列满了就返回 'overloaded' 错误。"

"优化手段：加索引让查询更快、用读副本分担读负载、把大用户拆到不同数据库（水平分库）。"

3. 查询性能是关键

"查询越快，吞吐量越高。一个没有索引的 SELECT * FROM users WHERE email = ? 要扫描全表——表里有 5000 行就是 5000 行读，消耗 5000 行的读额度。加上索引后，只需要读几行。"

"记住：永远给 WHERE 条件和 JOIN 字段加索引。"

4. 没有出口流量费

"D1 不收数据传输费。查询返回的数据不管多大，都不按流量计费——只按读取的行数计费。这比国内很多云数据库良心多了。"

免费额度详解

"最后看看大家最关心的免费额度："老张画了张表：

计费项	免费计划	付费计划 ($5/月)
存储	5 GB（总量）	前 5 GB 免费，之后 $0.75/GB-月
行读取	500 万行/天	前 250 亿行/月免费，之后 $0.001/百万行
行写入	10 万行/天	前 5000 万行/月免费，之后 $1.00/百万行

"几个关键概念要注意："

• 行读取：按查询扫描的行数计算，不是返回的行数。SELECT * FROM users 扫描 5000 行就是 5000 行读，即使你只需要 1 条。加索引能把扫描行数降到个位数。
• 行写入：INSERT 10 行算 10 行写入；UPDATE 影响多少行就算多少行写入。
• 免费额度每天 UTC 00:00 重置——北京时间早上 8 点。
• 读副本不额外收费——无论你开了几个副本，计费方式和没开一样。
• 空数据库也占存储：一个空数据库大约占 12 KB，一个空表占几 KB。虽然不多，但创建了几十个空数据库还是会有影响。

"500 万行读/天听着多，但如果不加索引，一个全表扫描就可能消耗几千行额度。索引不是可选项，是必选项。"

---

小结预告

本篇知识点回顾

知识点	核心内容
SQLite 特点	嵌入式、无独立进程、文件级数据库、读取快、零配置
SQLite 适合边缘的原因	无网络开销、毫秒级启动、资源占用极少，契合边缘按需使用模式
D1 架构	一个写主库 (Primary) + 多个全球读副本 (Read Replicas)
主库 (Primary)	唯一写入入口，所有写操作路由到此，保证写入顺序性
读副本 (Read Replica)	分布在全球区域，只处理读查询，降低读取延迟
异步复制	主库写入后异步同步到副本，副本可能存在数据滞后
顺序一致性	通过 bookmark 机制保证读己之写、单调读，比最终一致性更强
单实例单线程	每个 D1 实例顺序处理查询，吞吐量取决于查询速度
创建数据库	npx wrangler d1 create <name>
绑定到 Worker	wrangler.jsonc 中配置 d1_databases，代码中 env.BINDING_NAME
执行 SQL	wrangler d1 execute <db> --local/--remote --file/--command
--local vs --remote	local 操作本地开发库，remote 操作线上生产库
prepare()	创建预编译语句，用 ? 占位符，防止 SQL 注入
bind()	安全绑定参数值到占位符
all()	返回所有匹配行，适合查询多行
first()	返回第一行或 null，适合查询单行
run()	执行写操作，返回 meta 信息（含 last_row_id）
batch()	在单个事务中执行多条语句，全成功或全回滚
raw()	返回数组的数组，无字段名，高性能批量读取
Sessions API	withSession() 启用读副本路由，bookmark 保证一致性
索引的重要性	索引大幅减少扫描行数，降低延迟和额度消耗
免费额度	5GB 存储 + 500 万行读/天 + 10 万行写/天
读副本免费	启用读副本不额外收费，计费方式不变
单库最大 500MB	免费计划单库 500MB，付费计划 10GB，按需水平拆分

动手挑战

1. 基础挑战：创建一个 D1 数据库，建一张 users 表，通过 wrangler d1 execute 命令行完成插入和查询。然后在 Worker 中绑定 D1，实现 GET /api/users 返回所有用户列表。

2. 进阶挑战：给星火AI 实现完整的对话记录 API——POST /api/conversations 写入对话，GET /api/conversations?uid=1&page=1&size=20 分页查询。注意给 user_id 和 created_at 加复合索引。再写一个月度排行榜聚合查询，统计本月生成图片最多的前 10 名用户。

3. 折腾挑战：启用 D1 读副本，用 Sessions API 重写查询代码。尝试用 withSession("first-primary") 实现"用户提交反馈后立刻能看到自己的反馈"的场景——写入后第一个查询走主库确保读到最新数据，后续查询走读副本降低延迟。思考：如果不加 bookmark，可能出现什么问题？

提示：每个查询返回的 meta 对象里有 rows_read 和 rows_written 字段，可以用来监控额度消耗。在 Dashboard 的 D1 页面 → Metrics → Row Metrics 也能看到用量趋势。如果你的 rows_read 异常高，大概率是有查询没加索引——去查慢查询日志。

下回预告

林一把用户注册、对话记录、图片元数据全搬进了 D1 数据库，SQL 查询跑得飞快，分页、排序、聚合统计一气呵成。小王测了一遍功能，满意地点了点头。

"不过林一，"小王突然想起什么，"我刚才用脚本测了一下注册接口——每秒能发 50 个请求。你猜怎么着？1 分钟灌进来了 3000 个假用户。"

林一一看日志，全是 test_001、test_002、test_003……同一个 IP，疯狂调用 /api/users/register。

"得给注册接口加个验证——区分人和机器人。"

老张在旁边插话："Cloudflare 有个现成的方案——Turnstile。不需要用户找红绿灯、找斑马线，无感验证，免费使用。"

"无感验证？用户什么都不用做？"

"对，Turnstile 会通过浏览器行为特征自动判断。下一篇文章，给星火AI 的门口装个安检门。"

下篇预告：《数据库搞好了，但注册接口被机器人疯狂刷……——Turnstile 验证码》