Cloudflare D1：免费 5GB SQLite，全球分布

数据库这件小事

独立开发者最头疼的事之一就是数据库。

租个 RDS？最便宜的 MySQL 也要 $10/月起，还没算存储和流量。自己装？要操心备份、升级、主从复制、连接池……Supabase 和 PlanetScale 很好，但免费套餐一过期就开始烧钱。

D1 是 Cloudflare 的 Serverless 数据库，基于 SQLite。免费 5GB 存储、每天 500 万次读、10 万次写，永久免费。付费后 $0.001/百万次读，存储 $0.75/GB-月，而且出口流量完全免费。

对独立开发者来说，这意味着你可以在 $0 的成本下跑一个正经的数据库，等到真正有用户了再按量付费。

• • 免费额度大：5GB 存储 + 500 万行读/天，永久免费
• • SQLite 兼容：标准 SQL 语法，迁移零成本
• • 全球读副本：自动跨区域复制，延迟低
• • 零出口费：数据传出完全免费
• • 与 Workers 原生集成：一个 Binding 搞定，不需要连接字符串

D1 是什么？

D1 是一个 Serverless SQL 数据库，底层基于 SQLite。你可以把它理解为"不用运维的 SQLite"——你写标准的 SQL，D1 帮你处理存储、备份、复制、扩展。

和传统数据库最大的区别是：没有连接池的概念。D1 通过 Workers Binding 直接调用，每次查询都是一个独立的请求，不需要维护数据库连接。这意味着你不用担心连接数耗尽、连接泄漏这些问题。

D1 的架构：

用户请求    ↓┌─────────────┐│   Worker    │  ← 你的业务逻辑└─────────────┘    ↓ (Binding)┌─────────────┐│     D1      │  ← Serverless 数据库│  ┌───────┐  ││  │ Primary│  │  ← 写入节点│  └───┬───┘  ││      ↓      ││ ┌─────────┐ ││ │Replicas │ │  ← 全球只读副本（ENAM, WNAM, WEUR, EEUR, APAC）│ └─────────┘ │└─────────────┘

写入流程：Worker 通过 Binding 发起写请求 → D1 Primary 节点处理 → 返回结果。写入总是走 Primary 节点，保证强一致性。

读取流程：Worker 发起读请求 → D1 根据 Session 配置路由到最近的 Replica → 返回结果。使用 Sessions API 可以保证读到最新写入的数据（Sequential Consistency）。

D1 vs 其他数据库：到底差在哪？

关键差异：

1. 1. 免费额度最大方。D1 的 5GB 免费存储是 Supabase 的 10 倍、Neon 的 10 倍。读写操作的免费额度也远超同行。
2. 2. 零出口费。D1 的数据传出完全免费，而 Supabase 超出 5GB 后要收 $0.09/GB。
3. 3. SQLite vs PostgreSQL/MySQL。D1 用的是 SQLite 语法，对独立开发者来说更轻量。不需要管理用户、权限、Schema 这些复杂的东西。如果你的场景不需要存储过程、触发器、JSONB 高级查询，SQLite 完全够用。
4. 4. 原生 Workers 集成。不需要连接字符串、不需要连接池、不需要 VPC。一个 Binding 直接调用，延迟最低。
5. 5. 全球读副本免费。D1 的全球读副本不额外收费，而 PlanetScale 的多区域要额外付费。

免费额度 & 定价

免费额度（Workers Free 计划）

付费价格（Workers Paid，$5/月基础费）

关键点

• • 读写按行计费：不是按查询次数，是按扫描/影响的行数。一个 SELECT * FROM users 如果返回 100 行，就算 100 行读。
• • 写入比读取贵 1000 倍：写操作 $1.00/百万行 vs 读操作 $0.001/百万行，所以要控制写入频率。
• • 免费额度永久有效：不像 Supabase 的免费额度 12 个月后过期。
• • 存储按月计费：不是按天，是按整月的平均使用量。

5 分钟快速开始

方式一：Dashboard 创建

1. 1. 登录 Cloudflare Dashboard^[1]
2. 2. 左侧菜单选择 Storage & databases > D1
3. 3. 点击 Create database
4. 4. 输入数据库名称，选择地区
5. 5. 完成

方式二：Wrangler CLI

# 安装 wrangler（如果没有）npm install -g wrangler# 登录wrangler login# 创建数据库wrangler d1 create my-database# 输出类似：# ✅ Successfully created DB 'my-database'# [[d1_databases]]# binding = "DB"# database_name = "my-database"# database_id = "xxxx-xxxx-xxxx"# 执行 SQLwrangler d1 execute my-database --command "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)"# 执行 SQL 文件wrangler d1 execute my-database --file=./schema.sql# 查询数据wrangler d1 execute my-database --command "SELECT * FROM users"

方式三：在 Dashboard 中直接操作

创建数据库后，可以在 Dashboard 的 D1 页面直接执行 SQL、查看数据、导出备份。适合快速调试。

Workers 集成

D1 和 Workers 是天然搭档。通过 Binding，你可以在 Worker 中直接执行 SQL，零配置、低延迟。

配置 Binding

在 wrangler.toml 中添加：

[[d1_databases]]binding = "DB"database_name = "my-database"database_id = "xxxx-xxxx-xxxx"

或在 wrangler.jsonc 中：

{  "d1_databases": [    {      "binding": "DB",      "database_name": "my-database",      "database_id": "xxxx-xxxx-xxxx"    }  ]}

基础用法

interface Env {  DB: D1Database;}export default {  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise {    const url = new URL(request.url);    // 查询所有用户    if (request.method === "GET" && url.pathname === "/api/users") {      const { results } = await env.DB.prepare(        "SELECT * FROM users ORDER BY id DESC"      ).all();      return Response.json(results);    }    // 查询单个用户    if (request.method === "GET" && url.pathname.startsWith("/api/users/")) {      const id = url.pathname.slice(12);      const user = await env.DB.prepare(        "SELECT * FROM users WHERE id = ?"      ).bind(id).first();      if (!user) return new Response("Not Found", { status: 404 });      return Response.json(user);    }    // 创建用户    if (request.method === "POST" && url.pathname === "/api/users") {      const { name, email } = await request.json() as { name: string; email: string };      const { success, meta } = await env.DB.prepare(        "INSERT INTO users (name, email) VALUES (?, ?)"      ).bind(name, email).run();      return Response.json({ success, id: meta.last_row_id });    }    // 更新用户    if (request.method === "PUT" && url.pathname.startsWith("/api/users/")) {      const id = url.pathname.slice(12);      const { name, email } = await request.json() as { name: string; email: string };      await env.DB.prepare(        "UPDATE users SET name = ?, email = ? WHERE id = ?"      ).bind(name, email, id).run();      return Response.json({ success: true });    }    // 删除用户    if (request.method === "DELETE" && url.pathname.startsWith("/api/users/")) {      const id = url.pathname.slice(12);      await env.DB.prepare("DELETE FROM users WHERE id = ?").bind(id).run();      return Response.json({ success: true });    }    return new Response("Not Found", { status: 404 });  },};

Batch 批量操作

一次 Worker 调用中执行多条 SQL，减少往返次数：

const results = await env.DB.batch([  env.DB.prepare("INSERT INTO users (name) VALUES (?)").bind("Alice"),  env.DB.prepare("INSERT INTO users (name) VALUES (?)").bind("Bob"),  env.DB.prepare("INSERT INTO logs (action) VALUES (?)").bind("created_users"),]);

TypeScript 类型支持

D1 支持泛型，查询结果可以自动推导类型：

interface User {  id: number;  name: string;  email: string;  created_at: string;}const { results } = await env.DB.prepare(  "SELECT * FROM users WHERE id = ?").bind(id).all();// results 的类型是 User[]

实战场景

场景 1：用户系统

一个完整的用户注册、登录、信息管理：

interface Env {  DB: D1Database;}export default {  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise {    const url = new URL(request.url);    // 注册    if (request.method === "POST" && url.pathname === "/api/register") {      const { username, email, password } = await request.json() as {        username: string; email: string; password: string;      };      // 检查是否已存在      const existing = await env.DB.prepare(        "SELECT id FROM users WHERE email = ?"      ).bind(email).first();      if (existing) {        return Response.json({ error: "Email already exists" }, { status: 409 });      }      // 哈希密码（实际项目中用 bcrypt）      const passwordHash = await hashPassword(password);      const { meta } = await env.DB.prepare(        "INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)"      ).bind(username, email, passwordHash).run();      return Response.json({ success: true, id: meta.last_row_id });    }    // 登录    if (request.method === "POST" && url.pathname === "/api/login") {      const { email, password } = await request.json() as {        email: string; password: string;      };      const user = await env.DB.prepare(        "SELECT id, username, password_hash FROM users WHERE email = ?"      ).bind(email).first();      if (!user || !await verifyPassword(password, user.password_hash as string)) {        return Response.json({ error: "Invalid credentials" }, { status: 401 });      }      // 生成 token（实际项目中用 JWT）      const token = await generateToken(user.id as number);      return Response.json({ token, user: { id: user.id, username: user.username } });    }    // 获取用户信息    if (request.method === "GET" && url.pathname === "/api/me") {      const userId = await authenticate(request);      if (!userId) return new Response("Unauthorized", { status: 401 });      const user = await env.DB.prepare(        "SELECT id, username, email, created_at FROM users WHERE id = ?"      ).bind(userId).first();      return Response.json(user);    }    return new Response("Not Found", { status: 404 });  },};

对应的 Schema：

CREATE TABLE users (  id INTEGER PRIMARY KEY AUTOINCREMENT,  username TEXT NOT NULL,  email TEXT NOT NULL UNIQUE,  password_hash TEXT NOT NULL,  created_at TEXT DEFAULT (datetime('now')));CREATE INDEX idx_users_email ON users(email);

场景 2：博客系统

文章 + 评论 + 标签，典型的 CMS 场景：

interface Env {  DB: D1Database;}export default {  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise {    const url = new URL(request.url);    // 发布文章    if (request.method === "POST" && url.pathname === "/api/posts") {      const { title, content, tags } = await request.json() as {        title: string; content: string; tags: string[];      };      const { meta } = await env.DB.prepare(        "INSERT INTO posts (title, content, status) VALUES (?, ?, 'published')"      ).bind(title, content).run();      const postId = meta.last_row_id;      // 批量插入标签      if (tags.length > 0) {        const stmts = tags.map((tag) =>          env.DB.prepare(            "INSERT INTO post_tags (post_id, tag) VALUES (?, ?)"          ).bind(postId, tag)        );        await env.DB.batch(stmts);      }      return Response.json({ success: true, id: postId });    }    // 获取文章列表（带分页）    if (request.method === "GET" && url.pathname === "/api/posts") {      const page = parseInt(url.searchParams.get("page") || "1");      const limit = parseInt(url.searchParams.get("limit") || "10");      const offset = (page - 1) * limit;      const { results } = await env.DB.prepare(        "SELECT id, title, created_at FROM posts WHERE status = 'published' ORDER BY created_at DESC LIMIT ? OFFSET ?"      ).bind(limit, offset).all();      const total = await env.DB.prepare(        "SELECT COUNT(*) as count FROM posts WHERE status = 'published'"      ).first();      return Response.json({        posts: results,        pagination: { page, limit, total: total?.count },      });    }    // 获取文章详情（带标签）    if (request.method === "GET" && url.pathname.startsWith("/api/posts/")) {      const id = url.pathname.slice(11);      const post = await env.DB.prepare(        "SELECT * FROM posts WHERE id = ?"      ).bind(id).first();      if (!post) return new Response("Not Found", { status: 404 });      const { results: tags } = await env.DB.prepare(        "SELECT tag FROM post_tags WHERE post_id = ?"      ).bind(id).all();      return Response.json({ ...post, tags: tags.map((t) => t.tag) });    }    return new Response("Not Found", { status: 404 });  },};

对应的 Schema：

CREATE TABLE posts (  id INTEGER PRIMARY KEY AUTOINCREMENT,  title TEXT NOT NULL,  content TEXT NOT NULL,  status TEXT DEFAULT 'draft',  created_at TEXT DEFAULT (datetime('now')),  updated_at TEXT DEFAULT (datetime('now')));CREATE TABLE post_tags (  id INTEGER PRIMARY KEY AUTOINCREMENT,  post_id INTEGER NOT NULL,  tag TEXT NOT NULL,  FOREIGN KEY (post_id) REFERENCES posts(id) ON DELETE CASCADE);CREATE INDEX idx_posts_status ON posts(status);CREATE INDEX idx_post_tags_post_id ON post_tags(post_id);CREATE INDEX idx_post_tags_tag ON post_tags(tag);

场景 3：Session 管理

D1 特别适合存储用户会话，配合 Workers 实现无状态认证：

interface Env {  DB: D1Database;}export default {  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise {    // 创建 Session    async function createSession(userId: number): Promise {      const token = crypto.randomUUID();      const expiresAt = new Date(Date.now() + 7 * 24 * 60 * 60 * 1000).toISOString();      await env.DB.prepare(        "INSERT INTO sessions (token, user_id, expires_at) VALUES (?, ?, ?)"      ).bind(token, userId, expiresAt).run();      return token;    }    // 验证 Session    async function getSession(token: string) {      return await env.DB.prepare(        "SELECT user_id, expires_at FROM sessions WHERE token = ? AND expires_at > datetime('now')"      ).bind(token).first();    }    // 清理过期 Session    async function cleanExpiredSessions() {      await env.DB.prepare(        "DELETE FROM sessions WHERE expires_at < datetime('now')"      ).run();    }    // ... 处理请求  },};

成本计算器

案例 1：个人博客

• • 100 篇文章，每篇 2000 字
• • 每天 5000 次页面访问

案例 2：SaaS 后台

• • 1 万用户，每人 10 条记录
• • 每天 10 万次 API 请求

案例 3：中型应用

• • 10 万用户，每人 100 条记录
• • 每天 100 万次 API 请求

对比 Supabase：同样场景，Supabase 免费套餐（500MB）早就超了，Pro 计划 $25/月起。

最佳实践 & 避坑指南

1. 用 Batch 减少往返

一次 Worker 调用中执行多条 SQL，用 DB.batch() 而不是多次 prepare().run()：

// 好：一次 batchawait env.DB.batch([  env.DB.prepare("INSERT INTO users (name) VALUES (?)").bind("Alice"),  env.DB.prepare("INSERT INTO logs (action) VALUES (?)").bind("user_created"),]);// 不好：多次调用await env.DB.prepare("INSERT INTO users (name) VALUES (?)").bind("Alice").run();await env.DB.prepare("INSERT INTO logs (action) VALUES (?)").bind("user_created").run();

2. 用 .first() 只取一条

如果你只需要一条结果，用 .first() 而不是 .all()：

// 好：只取一条const user = await env.DB.prepare(  "SELECT * FROM users WHERE id = ?").bind(id).first();// 不好：取所有再取第一条const { results } = await env.DB.prepare(  "SELECT * FROM users WHERE id = ?").bind(id).all();const user = results[0];

3. 用 .raw() 减少序列化开销

如果你需要高性能，用 .raw() 获取原始数组而不是对象：

// 原始数组：[["Alice", "alice@example.com"], ["Bob", "bob@example.com"]]const rows = await env.DB.prepare("SELECT name, email FROM users").raw();

4. 注意行数计费方式

D1 按扫描/影响的行数计费，不是按查询次数。一个全表扫描会消耗所有行的读额度：

// 好：带索引的精确查询await env.DB.prepare("SELECT * FROM users WHERE email = ?").bind(email).first();// 只消耗 1 行读// 不好：全表扫描await env.DB.prepare("SELECT * FROM users").all();// 消耗所有行的读

5. 合理设计 Schema

• • 为常用查询字段添加索引
• • 避免在 D1 中存储大文件（最大 2MB/行），文件存 R2
• • 用 INTEGER 而不是 TEXT 存储时间戳，查询更快

-- 好：索引 + 合理类型CREATE INDEX idx_posts_created_at ON posts(created_at);-- 不好：无索引 + TEXT 时间-- 每次 ORDER BY created_at 都是全表扫描

6. 用 Sessions API 保证一致性

D1 有全球读副本，但读副本有延迟。如果你先写后读，要用 Sessions API：

// 创建 session，保证读到最新数据const session = env.DB.withSession("first-primary");// 写入await session.prepare("INSERT INTO users (name) VALUES (?)").bind("Alice").run();// 读取（保证读到刚才的写入）const user = await session.prepare("SELECT * FROM users WHERE name = ?").bind("Alice").first();

7. 避坑

• • 不要在 D1 中存储文件：大文件（图片、视频）存 R2，D1 只存元数据
• • 注意免费额度的每日限制：500 万行读/天、10 万行写/天，超出会报错
• • 单库最大容量：免费 500MB，付费 10GB，超出需要拆分
• • D1 是单线程：每秒约 1000 次查询（假设每次 1ms），高并发场景要考虑缓存
• • 不要用 D1 做实时计数器：写入频率太高会快速消耗额度，用 KV 或 Durable Objects

相关资源

https://developers.cloudflare.com/d1/

引用链接

[1] Cloudflare Dashboard: https://dash.cloudflare.com/