后端缓存抽象【免费下载链接】dataloaderDataLoader is a generic utility to be used as part of your applications data fetching layer to provide a consistent API over various backends and reduce requests to those backends via batching and caching.项目地址https://gitcode.com/gh_mirrors/da/dataloader点击查看免费下载DataLoader 通常被看作键值存储key-value store的最佳搭档但 SQL 数据库同样存在天然的批量查询机制——WHERE IN子句。本文以 examples/SQL.md 为骨架完整讲解如何用 DataLoader 配合 SQLite 实现一次事件循环内多次load()合并为一条SELECT ... WHERE id IN (...)的批量加载模式并深入源码剖析其排序约束、缺失行处理、缓存失效与maxBatchSize控制等底层原理。读完本文你将能够在自己的 Node.js 服务中为任意 SQL 表写出正确、可复用、可接入 GraphQL 的 DataLoader 查询层。为什么 SQL 也需要 DataLoaderDataLoader 的核心价值在于批处理batching与缓存caching。虽然它天然适合 RedisMGET这类命令式键值存储但对于 SQL 数据库SELECT * WHERE IN语句本身就提供了等价的批量能力因此只要查询保持简单例如按主键取整行DataLoader 完全可以胜任 SQL 场景。原示例文档给出的典型用法是以行主键id为 key请求整行数据。实际项目中你完全可以根据业务调整——例如只取某些列、按username查询等只要 batch 函数batch load function满足 DataLoader 的约束即可。// 安装依赖 npm install --save dataloader sqlite3完整示例基于 SQLite 的 DataLoader 实战以下是 examples/SQL.md 的核心示例它通过sqlite3驱动打开一个本地数据库文件并构造userLoaderconst DataLoader require(dataloader); const sqlite3 require(sqlite3); const db new sqlite3.Database(./to/your/db.sql); // 派发一条 WHERE-IN 查询并确保响应中行顺序正确。 const userLoader new DataLoader( ids new Promise((resolve, reject) { db.all( SELECT * FROM users WHERE id IN $ids, { $ids: ids }, (error, rows) { if (error) { reject(error); } else { resolve( ids.map( id rows.find(row row.id id) || new Error(Row not found: ${id}), ), ); } }, ); }), ); // 使用方式 const promise1 userLoader.load(1234); const promise2 userLoader.load(5678); const [user1, user2] await Promise.all([promise1, promise2]); console.log(user1, user2);这段代码的核心流程分四步构造 loadernew DataLoader(batchFn)其中batchFn接收一个 key 数组此处为id数组。派发批量查询在batchFn内部执行db.all(SELECT * FROM users WHERE id IN $ids, { $ids: ids }, ...)。sqlite3会把$ids参数展开为以逗号分隔的占位符从而将按多个 id 查整行合并为一条SQL 语句。重组结果顺序在回调中执行ids.map(id rows.find(row row.id id) || new Error(...))把数据库返回的行按请求 id 的顺序排列。消费结果load(1234)与load(5678)在同一事件循环帧tick内并发发出Promise.all同时拿到两行数据。关键设计点一结果顺序必须与请求 key 顺序严格对齐DataLoader 对 batch 函数有两个硬性约束见 README.md 的 Batch Function 一节以及 src/index.js 中对返回值长度的断言返回的 values 数组长度必须等于传入的 keys 数组长度values 中每个下标必须与 keys 中相同下标的 key 一一对应。这是因为 DataLoader 在派发完成后会按下标把 values 依次 resolve 给对应的load()调用。源码中的校验逻辑如下src/index.jsif (values.length ! batch.keys.length) { throw new TypeError( DataLoader ... did not return a Promise of an Array of the same length as the Array of keys., ); } // 逐一下标 resolve / reject for (let i 0; i batch.callbacks.length; i) { const value values[i]; if (value instanceof Error) { batch.callbacks[i].reject(value); } else { batch.callbacks[i].resolve(value); } }而 SQL 数据库并不保证WHERE IN的返回顺序与IN列表一致。设想请求 keys 为[2, 9, 6, 1]后端却返回了{ id: 9, name: Chicago } { id: 1, name: New York } { id: 2, name: San Francisco }若直接透传下标错位会导致load(2)拿到 id 为 9 的 Chicago 行。因此示例中通过ids.map(id rows.find(row row.id id) || ...)将结果重排为与 keys 对齐的顺序[ { id: 2, name: San Francisco }, { id: 9, name: Chicago }, null, // 或 new Error() { id: 1, name: New York }, ];实战提示数据量较大时rows.find是 O(n²) 开销可在 batch 函数内先构建Map(id → row)再按 ids 顺序取值效果等价且更快。关键设计点二用 Error 实例表达行不存在SQL 批量查询可能因部分 id 无对应行而返回更少的行。示例中的|| new Error(Row not found: ${id})有两层含义补齐数组长度确保 values 长度与 keys 长度一致满足 DataLoader 的约束语义化失败源码中dispatchBatch会检查每个 value 是否为Error实例若是则reject对应的 Promisesrc/index.js因此load(missing-id)会抛错而不是静默返回null。需要留意的是缓存行为差异README Caching Errors 一节整个 batch 失败函数抛错或返回 rejected Promise时不会缓存但单个 value 是 Error 实例时该 Error 会被缓存以避免反复加载同样的错误。若你希望行不存在的错误不要被缓存例如数据可能很快补上可以这样处理try { const user await userLoader.load(missing-id); } catch (error) { if (/* 判断该错误不应被缓存 */) { userLoader.clear(missing-id); } throw error; }批量机制原理为什么同帧内的 load 会被合并成一条 SQLDataLoader 默认把同一帧执行一次事件循环 tick内的所有load()合并为一次 batch 调用。其调度核心是getCurrentBatchsrc/index.jsloader 持有一个未派发的 batch新 key 不断追加其中直到超过maxBatchSize或调度器触发派发。派发时机由batchScheduleFn决定默认实现是enqueuePostPromiseJobsrc/index.js。它在 Node.js 环境下通过Promise.resolve().then(() process.nextTick(fn))巧妙地把 batch 派发排队到当前执行帧以及所有 Promise 微任务PromiseJobs刷新之后const enqueuePostPromiseJob typeof process object typeof process.nextTick function ? function (fn) { if (!resolvedPromise) resolvedPromise Promise.resolve(); resolvedPromise.then(() { process.nextTick(fn); }); } : typeof setImmediate function ? function (fn) { setImmediate(fn); } : function (fn) { setTimeout(fn); };这意味着即使你的代码在多个 Promise 链中先后调用load()只要它们都发生在同一帧的微任务刷新窗口内仍然会被合并。在 SQLite 示例中这等价于原本 N 次SELECT * FROM users WHERE id ?的往返被压缩为一条WHERE id IN (...)查询。若你想手动控制派发时机比如收集 100ms 窗口内的请求或手动 dispatch可以传入自定义batchScheduleFnconst myLoader new DataLoader(myBatchFn, { batchScheduleFn: callback setTimeout(callback, 100), });缓存与失效SQL 场景下的正确姿势按请求per-request创建实例DataLoader 的缓存是进程内 memoization 缓存不能替代 Redis / Memcache 等共享应用级缓存它只服务于单次请求内不重复加载同一数据。因此禁止让多个不同权限的用户共享同一个 loader 实例典型做法是每个 HTTP 请求创建一组新 loaderREADME Caching Per-Request 一节function createLoaders(authToken) { return { users: new DataLoader(ids genUsers(authToken, ids)), cdnUrls: new DataLoader(rawUrls genCdnUrls(authToken, rawUrls)), stories: new DataLoader(keys genStories(authToken, keys)), }; } // 处理每个 Web 请求时 const loaders createLoaders(request.query.authToken); const user await loaders.users.load(4);写入/更新后必须 clearSQL 场景最常见的缓存失效时机是同一次请求内发生了写入。README 用一条 SQL UPDATE 演示了标准流程// 请求开始... const userLoader new DataLoader(...); // 某个值被加载并进入缓存 const user await userLoader.load(4); // 发生变更缓存可能已过期 await sqlRun(UPDATE users WHERE id4 SET usernamezuck); userLoader.clear(4); // 再次加载拿到最新数据 const user await userLoader.load(4); // 请求结束clear(key)与clearAll()都返回 loader 自身以支持链式调用见 src/index.js。此外prime(key, value)可以预填缓存常用于用 id 加载后顺便填充 username 索引的双向加载模式README Loading by alternative keys 一节。关闭缓存与自定义缓存new DataLoader(myBatchFn, { cache: false })每次load()都产生新 Promise且批函数可能收到含重复 key 的数组每个load()调用对应一个 key 实例批函数必须为每个重复 key 提供值长生命周期 loader 建议提供自定义cacheMap实现get/set/delete/clear四个方法即可src/index.js 定义了其类型约束例如用 LRU 限制内存占用cacheKeyFn可自定义缓存键生成方式默认key key适合对象 key 的等价判断name选项可为 loader 命名便于 APM 工具观测。缓存命中与批处理并行不悖缓存命中的 key不会出现在批函数的 keys 中但其 Promise 仍会等待当前 batch 完成后再一起 resolvesrc/index.js 中cacheHits的处理逻辑。这使得后续依赖加载如user.bestFriendID能与未命中的请求在同一帧内再次合并README 中的prime(1, { bestFriend: 3 })示例正是利用这一特性把 3 次请求压缩为 2 次。用 maxBatchSize 控制单条 SQL 的查询规模DataLoader 的完整选项见 README.md 的 API 表格如下Option Key类型默认值说明batchBooleantrue设为false即禁用批处理等价于maxBatchSize: 1maxBatchSizeNumberInfinity限制传入 batch 函数的 key 数量上限设为1即禁用批处理batchScheduleFnFunction默认调度器自定义批量派发调度函数cacheBooleantrue设为false禁用 memoization 缓存等价于cacheMap: nullcacheKeyFnFunctionkey key由加载 key 生成缓存 keycacheMapObjectnew Map()自定义缓存实例可为nullnameStringnull实例名称供 APM 使用其中maxBatchSize对 SQL 场景尤其实用数据库驱动如 SQLite 的参数绑定、MySQL 的max_allowed_packet对单条IN子句可容纳的占位符数量有上限业务上也往往需要控制单次查询的数据量。源码getValidMaxBatchSizesrc/index.js要求其为不小于 1 的数字否则抛TypeErrorconst myLoader new DataLoader(ids myBatchQuery(ids), { maxBatchSize: 100, // 每条 WHERE IN 最多包含 100 个 id });当 key 数超过上限时getCurrentBatch会另起新 batch从而把一次大查询切分为多条受控的 SQL。更多 SQL 变体Knex.js 的 whereIn如果不想手写 SQL可以参考仓库中的 examples/Knex.md借助 Knex 查询构造器的.whereIn(id, ids)在保留同样按 ids 重排结果逻辑的同时免去手写 SQLconst loaders { user: new DataLoader(ids db .table(users) .whereIn(id, ids) .select() .then(rows ids.map(id rows.find(x x.id id))), ), story: new DataLoader(ids db .table(stories) .whereIn(id, ids) .select() .then(rows ids.map(id rows.find(x x.id id))), ), // 一对多按 author_id 批量取 stories storiesByUserId: new DataLoader(ids db .table(stories) .whereIn(author_id, ids) .select() .then(rows ids.map(id rows.filter(x x.author_id id))), ), };注意第三个 loader 返回的是每个 key 对应一个数组同样满足 values 与 keys 长度一致、下标对齐的约束——可见一行对一行并非唯一形态一对多同样可行。与 GraphQL 结合把 SQLite loader 接入 User 类型DataLoader 最典型的落地场景是 GraphQL 服务。README 的 Using with GraphQL 一节以本 SQLite 示例为数据源定义了User类型bestFriend字段直接userLoader.load(user.bestFriendID)friends字段先经queryLoader执行按用户查好友 id 的 SQL再逐 iduserLoader.load(row.toID)const UserType new GraphQLObjectType({ name: User, fields: () ({ name: { type: GraphQLString }, bestFriend: { type: UserType, resolve: user userLoader.load(user.bestFriendID), }, friends: { args: { first: { type: GraphQLInt } }, type: new GraphQLList(UserType), resolve: async (user, { first }) { const rows await queryLoader.load([ SELECT toID FROM friends WHERE fromID? LIMIT ?, user.id, first, ]); return rows.map(row userLoader.load(row.toID)); }, }, }), });假设一个嵌套查询同时解析me、bestFriend、friends及其各自的bestFriend朴素实现最多可能发出 13 次数据库请求而接入上述 loader 后最多 4 次命中缓存时更少。这正是GraphQL 字段解析函数 DataLoader组合的价值所在。测试佐证仓库如何验证这些行为仓库的测试文件 src/tests/dataloader.test.js 为本文所述行为提供了直接依据同帧合并it(batches multiple requests)断言连续两次load(1)、load(2)后批函数只被调用一次且收到[1, 2]第 93-104 行maxBatchSize 切分it(batches multiple requests with max batch sizes)验证maxBatchSize: 2时 3 个 key 被拆为两批第 106-120 行loadMany 错误语义it(supports loading multiple keys in one call with errors)验证loadMany([a,b,bad])返回[a,b,Error]而不是整体 reject第 79-91 行缓存 API测试覆盖prime、clearAll、clear(key).prime(key, value)等链式行为。这意味着把 SQLite 示例中的Promise.all([load(1234), load(5678)])换成loadMany([1234,5678])同样安全——loadMany永远 resolve每个元素要么是值、要么是 Error 实例src/index.js。小结SQL 虽然不是键值存储但只要查询保持简单按主键取整行、按外键取子集WHERE IN就能与 DataLoader 的批处理模型完美契合。落地的关键纪律可以总结为四条结果必须按 keys 重排对齐、缺失行用 Error 表达并留意其缓存语义、每个请求新建 loader 实例并在写入后 clear、用 maxBatchSize 控制单条 SQL 规模。在此基础上无论是原生 SQLite、Knex 还是其他 SQL 驱动你都能构建出与 GraphQL 等上层框架无缝衔接的高效数据加载层。赞分享后端缓存抽象【免费下载链接】dataloaderDataLoader is a generic utility to be used as part of your applications data fetching layer to provide a consistent API over various backends and reduce requests to those backends via batching and caching.项目地址https://gitcode.com/gh_mirrors/da/dataloader点击查看免费下载相关推荐SQLModel 数据过滤指南使用 .where() 与 SQL WHERE 精准查询数据库行SQLModel 数据过滤指南使用 .where 与 SQL WHERE 精准查询数据库行 导读 本篇教程围绕 SQLModel 教程系列中的数据过滤章节ORM数据库后端快速解决ComfyUI-Impact-Pack边界框检测器参数错误完整指南快速解决ComfyUI Impact Pack边界框检测器参数错误完整指南 ComfyUI Impact Pack是ComfyUI的强大扩展包专门用于图像处AI 应用计算机视觉图像处理SQLite 关系数据库查询实战使用 VS Code 与 SQL 查询机场数据库Data-Science-For-Beginners 第 5 课实验SQLite 关系数据库查询实战使用 VS Code 与 SQL 查询机场数据库Data Science For Beginners 第 5 课实验 本指数据科学教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考