Cloudflare D1 高级优化：连接池管理、读写分离与大规模数据迁移

前置阅读：建议先阅读 Cloudflare D1 + Pages Functions 全栈实战。

关键概念：D1 在每个边缘节点运行 SQLite 实例，读操作本地响应，写操作通过 Durable Objects 强一致性协调。理解这一架构是优化的前提：“读本地，写全局”。

1. ² 连接管理与预处理

   D1 在 Workers 中通过绑定访问，无传统连接池概念，但预处理（Prepared Statements）仍然关键：

   // workers/db-optimized.ts
   export interface Env {
     DB: D1Database;
   }
   
   // ❌ 低效：每次字符串拼接 + 全量解析
   async function badQuery(env: Env, userId: string) {
     const result = await env.DB.prepare(
       `SELECT * FROM users WHERE id = '${userId}'`  // SQL 注入风险！
     ).all();
     return result;
   }
   
   // ✅ 高效：预编译 + 参数绑定 + 指定返回列
   async function goodQuery(env: Env, userId: string) {
     const result = await env.DB.prepare(
       `SELECT id, name, email, created_at FROM users WHERE id = ?`
     )
       .bind(userId)
       .first();
     return result;
   }
   
   // ✅ 批量插入：单事务多行（减少 WAL 写入）
   async function batchInsert(env: Env, users: User[]) {
     const placeholders = users.map(() => "(?, ?, ?)").join(", ");
     const flat = users.flatMap(u => [u.name, u.email, u.createdAt]);
   
     // D1 限制：单个 prepared statement 最多 100 个变量
     // 所以 users 最多 ~33 条
     const BATCH_LIMIT = 30;
     const batches = chunk(users, BATCH_LIMIT);
   
     const results = await Promise.all(
       batches.map(batch => {
         const ph = batch.map(() => "(?, ?, ?)").join(", ");
         const values = batch.flatMap(u => [u.name, u.email, u.createdAt]);
         return env.DB.prepare(
           `INSERT INTO users (name, email, created_at) VALUES ${ph}`
         )
           .bind(...values)
           .run();
       })
     );
   
     return results;
   }
   
   function chunk<T>(arr: T[], size: number): T[][] {
     return Array.from({ length: Math.ceil(arr.length / size) }, (_, i) =>
       arr.slice(i * size, i * size + size)
     );
   }
   

   D1 限制清单：

   限制	数值	优化策略
   单条 SQL 最大变量数	100	批量分批
   单条查询最大返回行	1,000	分页 + 游标
   数据库大小（目前）	500MB	分片 / 归档
   单次请求超时	30s	复杂查询拆分
   并发写入	单 DO 串行	队列化写入

2. ³ 查询计划分析（EXPLAIN QUERY PLAN）

   -- 分析查询性能
   EXPLAIN QUERY PLAN
   SELECT o.id, o.amount, u.name
   FROM orders o
   JOIN users u ON o.user_id = u.id
   WHERE o.created_at > '2024-01-01'
   ORDER BY o.amount DESC
   LIMIT 20;
   

   典型的输出解读：

   QUERY PLAN
   |--SEARCH orders USING INDEX idx_orders_created_at(created_at>?)
   |--SEARCH users USING INTEGER PRIMARY KEY (rowid=?)
   `--USE TEMP B-TREE FOR ORDER BY
   

   优化方向：

   问题信号	含义	修复
   SCAN TABLE	全表扫描	添加索引
   USE TEMP B-TREE FOR ORDER BY	内存排序	创建覆盖索引
   SEARCH ... USING INDEX	索引命中	良好
   嵌套循环 JOIN 过慢	驱动表过大	重写为子查询或分页

   -- 复合索引优化 ORDER BY + WHERE
   CREATE INDEX idx_orders_created_amount ON orders(created_at, amount DESC);
   -- 覆盖索引：包含查询所需全部列，避免回表
   CREATE INDEX idx_orders_covering
   ON orders(user_id, created_at, amount, id);
   

3. ⁴ 大规模数据导入

   百万级数据迁移的最佳实践：

   // scripts/migrate-to-d1.ts
   import { execSync } from "child_process";
   
   async function bulkImport(dumpFile: string, dbName: string) {
     // Step 1: 本地 SQLite 预处理（索引延后创建）
     const tempDb = "temp_migrate.db";
     execSync(`sqlite3 ${tempDb} < ${dumpFile}`);
   
     // Step 2: 导出为 D1 兼容的 SQL
     // D1 支持大部分 SQLite 语法，但有限制：
     // - 不支持 ALTER TABLE DROP COLUMN
     // - 外键默认关闭（PRAGMA foreign_keys = OFF）
     execSync(`sqlite3 ${tempDb} .dump > d1_dump.sql`);
   
     // Step 3: 分批写入 D1（每批 < 50MB SQL）
     const sql = readFileSync("d1_dump.sql", "utf-8");
     const batches = splitIntoBatches(sql, 10_000);  // 每批 10K 行
   
     for (let i = 0; i < batches.length; i++) {
       console.log(`Importing batch ${i + 1}/${batches.length}...`);
       await importBatch(dbName, batches[i]);
       await sleep(1000);  // 避免速率限制
     }
   
     // Step 4: 批量创建索引（导入完成后再建，加速写入）
     await createIndexes(dbName);
   }
   
   async function importBatch(dbName: string, sql: string) {
     const response = await fetch(
       `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/d1/database/${dbName}/query`,
       {
         method: "POST",
         headers: {
           Authorization: `Bearer ${API_TOKEN}`,
           "Content-Type": "application/json",
         },
         body: JSON.stringify({ sql }),
       }
     );
     return response.json();
   }
   

   wrangler 导入（简化版）：

   # 将本地 SQLite 转换为 D1 兼容格式
   wrangler d1 execute my-db --local --file=./schema.sql
   wrangler d1 execute my-db --local --file=./data.sql
   
   # 生产环境执行
   wrangler d1 execute my-db --remote --file=./schema.sql
   

4. ⁵ 读写分离架构

   D1 不支持原生读写分离，可在 Workers 层模拟：

   // workers/read-write-split.ts
   export interface Env {
     DB_PRIMARY: D1Database;   // 写操作
     DB_REPLICA: D1Database;   // 读操作（通过 D1 的 read replication 或外部同步）
   }
   
   // 注：目前 D1 的 read replication 为自动行为（读就近），
   // 以下架构面向多数据库联邦场景
   
   export class DatabaseRouter {
     constructor(private env: Env) {}
   
     async query(sql: string, params?: any[], opts?: { readOnly?: boolean }) {
       if (opts?.readOnly) {
         // 读取操作：优先使用本地缓存 / 副本
         return this.env.DB_REPLICA.prepare(sql).bind(...(params || [])).all();
       }
       // 写入操作：必须走主库
       return this.env.DB_PRIMARY.prepare(sql).bind(...(params || [])).run();
     }
   }
   
   // 更实际的方案：CQRS 模式分离读写模型
   export async function handleCommand(command: Command, env: Env) {
     // 写模型：D1 事务处理
     const db = env.DB_PRIMARY;
     await db.prepare("BEGIN TRANSACTION").run();
     try {
       await db.prepare(command.sql).bind(...command.params).run();
       await db.prepare("COMMIT").run();
   
       // 异步通知读模型更新（Event-driven）
       await env.EVENT_QUEUE.send({ type: "ENTITY_UPDATED", data: command });
     } catch (e) {
       await db.prepare("ROLLBACK").run();
       throw e;
     }
   }
   

5. ⁶ D1 与外部数据库联邦查询

   // workers/federated-query.ts
   export async function federatedSearch(
     env: Env,
     userId: string
   ): Promise<UnifiedProfile> {
     // 并行查询多个数据源
     const [localProfile, remoteOrders, cachedPreferences] = await Promise.all([
       // D1：用户基础资料
       env.DB.prepare("SELECT * FROM users WHERE id = ?").bind(userId).first(),
   
       // Supabase/PostgreSQL：订单历史
       fetch(`${env.SUPABASE_URL}/rest/v1/orders?user_id=eq.${userId}`, {
         headers: { apikey: env.SUPABASE_KEY },
       }).then(r => r.json()),
   
       // KV：用户偏好缓存
       env.USER_PREFS.get(userId, { type: "json" }),
     ]);
   
     return {
       profile: localProfile,
       orders: remoteOrders,
       preferences: cachedPreferences || {},
     };
   }
   

   数据源	使用场景	延迟	一致性
   D1	用户资料、配置、关系数据	< 20ms	强一致（写）
   KV	会话、缓存、频率限制	< 1ms	最终一致
   R2	文件、日志、备份	< 50ms	最终一致
   外部 PostgreSQL	复杂分析、大数据量	> 100ms	依赖外部

6. ⁷ 性能基准与生产陷阱

   操作	1K 记录	100K 记录	1M 记录	建议
   单条 INSERT	50ms	-	-	用批量代替
   批量 INSERT (100条)	80ms	8s	-	每批 < 30 条
   SELECT（索引命中）	2ms	5ms	15ms	分页 + 索引
   SELECT（全表扫描）	5ms	200ms	3s	必须加索引
   JOIN（3表）	10ms	50ms	500ms	限制结果集大小

   常见陷阱：

   陷阱	现象	解决方案
   索引滥用	写入变慢，存储膨胀	只为 WHERE/JOIN/ORDER BY 列建索引
   大事务	写操作超时	拆分为小批次，中间 COMMIT
   缺少 LIMIT	返回行数超限（1,000）	总是加 LIMIT + OFFSET 游标
   N+1 查询	Workers 调用次数暴增	使用 JOIN 或 IN (…)

延伸阅读：

• Cloudflare D1 + Pages Functions 全栈实战 — 基础 CRUD 与 Drizzle ORM 使用
• Cloudflare KV 与 Durable Objects — 互补存储方案
• Cloudflare R2 对象存储实战 — 大数据文件处理