⚡

CloudflareD1数据库管理 cloudflare-d1

此技能涉及使用Cloudflare D1服务器无SQLite数据库进行边缘计算环境的数据库操作，包括创建、配置、迁移、查询优化和错误处理。关键词包括Cloudflare D1、SQLite、边缘数据库、Serverless、云数据库管理、无服务器计算。

Serverless 0 次安装 0 次浏览 更新于 3/8/2026

名称：cloudflare-d1 描述：Cloudflare D1 边缘无服务器 SQLite。用于数据库、迁移、绑定，或遇到 D1_ERROR、语句过长、请求队列过多错误。

关键词：d1, d1 数据库, cloudflare d1, wrangler d1, d1 迁移, d1 绑定, sqlite workers, 边缘数据库, d1 查询, sql cloudflare, 预处理语句, 批量查询, d1 api, wrangler 迁移, D1_ERROR, D1_EXEC_ERROR, 语句过长, 数据库绑定, sqlite cloudflare, sql workers api, d1 索引, 查询优化, d1 模式, 读取复制, 读取副本, withSession, Sessions API, 全局复制, 数据库复制, served_by_region, 书签, 顺序一致性 许可证：MIT 元数据： 版本：“3.0.0” 最后验证：“2025-01-15” 生产测试：true 令牌节省：“~58%” 错误预防：8 包含模板：3 包含参考：4 wrangler版本：“4.50.0” workers类型版本：“4.20251125.0” drizzle_orm版本：“0.44.7”

Cloudflare D1 数据库

状态：生产就绪 ✅ | 最后验证：2025-01-15

目录

  1. 什么是 D1？
  2. 快速开始
  3. 关键规则
  4. D1 API 方法
  5. 前 5 个用例
  6. 迁移最佳实践
  7. 常见模式
  8. SQLite 类型亲和性
  9. 前 5 个错误预防

什么是 D1？

Cloudflare D1 是边缘无服务器 SQLite：

  • 无服务器 SQL 数据库
  • 全球分布
  • 零冷启动
  • 标准 SQLite 语法
  • 读取复制以实现全局性能

🆕 2025 年新功能

D1 在 2025 年进行了重大更新：

性能（2025 年 1 月）

  • 全局延迟减少 40-60%（P50 查询时间）
  • 针对边缘执行优化的 SQLite 引擎
  • 对小于 100 MB 的数据库减少冷启动影响

可靠性（2025 年 9 月）

  • 自动查询重试：读取查询在瞬时故障时最多重试 2 次
  • 对应用程序代码透明（在 wrangler tail 中记录）

可扩展性（2025 年 4 月）

  • 读取复制（公共测试版）：全局部署读取副本
  • 读取密集型工作负载的读取吞吐量高达 2 倍
  • 用于读写分离的 Sessions API

合规性（2025 年 11 月）

  • 数据本地化：为 GDPR/数据主权指定欧盟/美国管辖区
  • 通过 --jurisdiction 标志或 wrangler.jsonc 配置

⚠️ 重大变更（2025 年 2 月 10 日）

  • 免费层硬限制强制执行：10 个数据库，每个 500 MB，每次调用 50 个查询
  • 超出限制 = 429 错误（之前仅为警告）
  • 行动：使用 wrangler d1 list 审查使用情况，必要时升级

完整详情：加载 references/2025-features.md

快速开始（5 分钟）

1. 创建数据库

bunx wrangler d1 create my-database

保存输出中的 database_id！

2. 配置绑定

添加到 wrangler.jsonc：

{
  "name": "my-worker",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "d1_databases": [
    {
      "binding": "DB",                    // env.DB
      "database_name": "my-database",
      "database_id": "<UUID>",
      "preview_database_id": "local-db"
    }
  ]
}

3. 创建迁移

bunx wrangler d1 migrations create my-database create_users

编辑 migrations/0001_create_users.sql：

CREATE TABLE IF NOT EXISTS users (
  user_id INTEGER PRIMARY KEY AUTOINCREMENT,
  email TEXT NOT NULL UNIQUE,
  username TEXT NOT NULL,
  created_at INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);

PRAGMA optimize;

4. 应用迁移

# 本地
bunx wrangler d1 migrations apply my-database --local

# 生产
bunx wrangler d1 migrations apply my-database --remote

5. 从 Worker 查询

import { Hono } from 'hono';

type Bindings = {
  DB: D1Database;
};

const app = new Hono<{ Bindings: Bindings }>();

app.get('/users/:email', async (c) => {
  const { results } = await c.env.DB.prepare(
    'SELECT * FROM users WHERE email = ?'
  )
    .bind(c.req.param('email'))
    .all();

  return c.json(results);
});

export default app;

加载 references/setup-guide.md 获取完整演练。

关键规则

始终做 ✅

  1. 使用预处理语句 与 .bind()（永远不要字符串连接）
  2. 为 WHERE/JOIN/ORDER BY 列创建索引
  3. 使用迁移 进行模式更改（永远不要手动 SQL）
  4. 批量查询 进行多个操作（.batch()）
  5. 模式更改后运行 PRAGMA optimize
  6. 显式处理错误（try/catch）
  7. 对时间戳使用 INTEGER（Date.now()）
  8. 部署迁移前本地测试
  9. 使用读取副本 实现全局读取性能
  10. SQL 查询前验证输入

永远不要做 ❌

  1. 永远不要将用户输入连接 到 SQL
  2. 永远不要将 database_id 提交 到公共仓库
  3. 永远不要跳过迁移 进行模式更改
  4. 永远不要使用 VARCHAR（使用 TEXT 代替）
  5. 永远不要跳过索引 用于过滤列
  6. 永远不要忽略 SQLite 类型亲和性规则
  7. **永远不要使用 SELECT *** 没有 LIMIT
  8. 永远不要运行迁移 而没有本地测试
  9. 永远不要超过 每行 1MB
  10. 永远不要使用 DATETIME（对时间戳使用 INTEGER）

D1 API 方法

prepare() - 执行查询

// 单结果
const { results } = await env.DB.prepare(
  'SELECT * FROM users WHERE email = ?'
)
  .bind(email)
  .all();

// 仅第一个结果
const user = await env.DB.prepare(
  'SELECT * FROM users WHERE user_id = ?'
)
  .bind(userId)
  .first();

// 原始结果（更快）
const { results } = await env.DB.prepare(
  'SELECT username FROM users'
)
  .raw();  // 返回数组而不是对象

batch() - 多个查询

const results = await env.DB.batch([
  env.DB.prepare('INSERT INTO users (email, username, created_at) VALUES (?, ?, ?)')
    .bind('user1@example.com', 'user1', Date.now()),
  env.DB.prepare('INSERT INTO users (email, username, created_at) VALUES (?, ?, ?)')
    .bind('user2@example.com', 'user2', Date.now()),
  env.DB.prepare('SELECT COUNT(*) as count FROM users')
]);

console.log('用户数:', results[2].results[0].count);

所有查询在单个事务中执行（全部成功或全部失败）。

exec() - 运行 SQL 字符串

// 仅用于迁移/设置
await env.DB.exec(`
  CREATE TABLE IF NOT EXISTS users (
    user_id INTEGER PRIMARY KEY,
    email TEXT NOT NULL
  );
  CREATE INDEX idx_email ON users(email);
`);

永远不要用于有用户输入的查询！

加载 references/query-patterns.md 获取完整 API 参考。

前 5 个用例

用例 1：用户 CRUD

// 创建
app.post('/users', async (c) => {
  const { email, username } = await c.req.json();

  const { results } = await c.env.DB.prepare(
    'INSERT INTO users (email, username, created_at) VALUES (?, ?, ?) RETURNING *'
  )
    .bind(email, username, Date.now())
    .all();

  return c.json(results[0]);
});

// 读取
app.get('/users/:id', async (c) => {
  const user = await c.env.DB.prepare(
    'SELECT * FROM users WHERE user_id = ?'
  )
    .bind(c.req.param('id'))
    .first();

  if (!user) {
    return c.json({ error: '未找到' }, 404);
  }

  return c.json(user);
});

// 更新
app.patch('/users/:id', async (c) => {
  const { username } = await c.req.json();

  await c.env.DB.prepare(
    'UPDATE users SET username = ?, updated_at = ? WHERE user_id = ?'
  )
    .bind(username, Date.now(), c.req.param('id'))
    .run();

  return c.json({ success: true });
});

// 删除
app.delete('/users/:id', async (c) => {
  await c.env.DB.prepare(
    'DELETE FROM users WHERE user_id = ?'
  )
    .bind(c.req.param('id'))
    .run();

  return c.json({ success: true });
});

用例 2：批量操作

app.post('/users/bulk', async (c) => {
  const users = await c.req.json();  // 用户数组

  const statements = users.map(user =>
    c.env.DB.prepare(
      'INSERT INTO users (email, username, created_at) VALUES (?, ?, ?)'
    ).bind(user.email, user.username, Date.now())
  );

  const results = await c.env.DB.batch(statements);

  return c.json({ inserted: results.length });
});

用例 3：读取复制（全局读取）

// 配置读取副本（任何区域）
const session = c.env.DB.withSession({
  preferredRegion: 'auto'  // 或 'weur', 'wnam', 'enam', 'apac'
});

// 从最近副本读取
const { results } = await session.prepare(
  'SELECT * FROM users WHERE email = ?'
)
  .bind(email)
  .all();

// 检查哪个区域服务请求
console.log('服务区域:', results[0].served_by_region);

加载 references/read-replication.md 获取完整指南。

用例 4：使用 Batch 的事务

// 用户间转移积分（原子性）
const results = await c.env.DB.batch([
  c.env.DB.prepare(
    'UPDATE users SET credits = credits - ? WHERE user_id = ?'
  ).bind(amount, fromUserId),
  c.env.DB.prepare(
    'UPDATE users SET credits = credits + ? WHERE user_id = ?'
  ).bind(amount, toUserId),
  c.env.DB.prepare(
    'INSERT INTO transactions (from_user, to_user, amount, created_at) VALUES (?, ?, ?, ?)'
  ).bind(fromUserId, toUserId, amount, Date.now())
]);

// 全部成功或全部失败（事务）

用例 5：分页

app.get('/users', async (c) => {
  const page = parseInt(c.req.query('page') || '1');
  const limit = 20;
  const offset = (page - 1) * limit;

  const { results } = await c.env.DB.prepare(
    'SELECT * FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?'
  )
    .bind(limit, offset)
    .all();

  return c.json({
    users: results,
    page,
    limit
  });
});

迁移最佳实践

1. 始终使用迁移

bunx wrangler d1 migrations create my-database add_users_avatar

2. 使迁移幂等

-- ✅ 好：幂等
CREATE TABLE IF NOT EXISTS users (...);
CREATE INDEX IF NOT EXISTS idx_email ON users(email);
DROP TABLE IF EXISTS old_table;

-- ❌ 坏：重运行时失败
CREATE TABLE users (...);
CREATE INDEX idx_email ON users(email);

3. 先本地测试

bunx wrangler d1 migrations apply my-database --local
bunx wrangler d1 execute my-database --local --command "SELECT * FROM users"

4. 添加 PRAGMA optimize

-- 迁移末尾
PRAGMA optimize;

加载 templates/schema-example.sql 获取完整模式模板。

何时加载参考

加载 references/setup-guide.md 当：

  • 首次 D1 设置
  • 创建第一个数据库
  • 配置绑定
  • 应用第一个迁移

加载 references/query-patterns.md 当：

  • 需要完整 API 参考
  • 复杂查询模式
  • 批量操作
  • 错误处理

加载 references/read-replication.md 当：

  • 设置全局读取
  • 需要全球低延迟
  • 理解 Sessions API
  • 需要顺序一致性

加载 references/best-practices.md 当：

  • 优化查询性能
  • 模式设计决策
  • 索引策略
  • 生产部署清单

加载 references/limits.md 当：

  • 遇到 429 错误或配额警告
  • 规划免费与付费层容量
  • 理解数据库/查询限制
  • 迁移到付费计划

加载 references/metrics-analytics.md 当：

  • 调查性能问题
  • 设置监控和警报
  • 使用 wrangler d1 insights 命令
  • 分析查询效率

加载 references/2025-features.md 当：

  • 从 v2.x 升级到 v3.x
  • 启用新功能（自动重试、管辖区、复制）
  • 理解重大变更（2025年2月10日强制执行）
  • 截止日期前迁移

交互式工具

代理（自主诊断）：

  • agents/d1-debugger.md：9阶段诊断（配置、迁移、查询、绑定、错误、限制、性能、时间旅行）
  • agents/d1-query-optimizer.md：性能分析（慢查询、缺失索引、优化建议）

命令（交互式向导）：

  • commands/cloudflare-d1:setup.md：交互式首次设置向导
  • commands/d1-create-migration.md：带验证的引导迁移创建

使用捆绑资源

参考 (references/)

模板 (templates/)

  • schema-example.sql - 带索引的完整模式
  • d1-worker-queries.ts - Worker 中所有查询模式
  • cloudflare-d1:setup-migration.sh - 完整设置脚本

常见模式

错误处理

try {
  const { results } = await env.DB.prepare(
    'SELECT * FROM users WHERE email = ?'
  )
    .bind(email)
    .all();

  return c.json(results);
} catch (error) {
  console.error('D1 错误：', error);
  return c.json({ error: '数据库错误' }, 500);
}

原始模式（性能）

// 返回数组而不是对象（更快）
const { results } = await env.DB.prepare(
  'SELECT user_id, email FROM users'
)
  .raw();

// results = [[1, 'user1@example.com'], [2, 'user2@example.com']]

COUNT 查询

const count = await env.DB.prepare(
  'SELECT COUNT(*) as count FROM users'
)
  .first('count');  // 获取单列值

console.log('总用户数：', count);

SQLite 类型亲和性

D1 使用 SQLite 类型亲和性：

声明类型 亲和性
INTEGER, INT INTEGER
TEXT, VARCHAR, CHAR TEXT
REAL, FLOAT, DOUBLE REAL
BLOB BLOB
（无类型） BLOB

最佳实践：

  • 对数字使用 INTEGER
  • 对字符串使用 TEXT（非 VARCHAR）
  • 对时间戳使用 INTEGER（Date.now()）
  • 对二进制数据使用 BLOB

前 5 个错误预防

  1. SQL 注入：使用 .bind()，永远不要字符串连接
  2. 缺失索引：为 WHERE/JOIN 列创建索引
  3. 迁移失败：先本地测试
  4. 类型混淆：对时间戳使用 INTEGER
  5. 批量大小：限制批次到 <500 条语句

加载 references/best-practices.md 获取完整错误预防。

官方文档

有问题？问题？

  1. 检查 references/setup-guide.md 获取设置
  2. 审查 references/query-patterns.md 获取 API 参考
  3. 查看 references/read-replication.md 获取全局读取
  4. 加载 references/best-practices.md 获取优化