批发工人
通过定义循环工人在自己的表格中加工批次的工作 自动的, 与排程, 解跳, 和恢复构建在.
你带来两个功能:
- A 类 工作查询 返回下一批工作,或明确闲置.
- A 类 工人突变 处理批量.
插入工作后,请拨 ping(...). 本部分负责其余部分:
- 每个工人一次运行一个循环.
- 支持拆弹,所以它们一起批量.
- 保持"温暖"的循环 短时间的投票冷却 所以一丝一毫的新工作 不打运行状态.
- 使用快照读取时会排出 这样同时插入不会引起OCC 重复一遍,在闲逛之前用真实的读数确认,这样就不会失去任何东西.
- 当队列排出后闲置, 下次自动重新启动 p.
- 监视循环和 如果它死了就重启它 (例如,意外事件) 错误),记录失败,以便您可以对此提醒.
这是组件背后的图案,比如 工作组合,提取后可以构建 。您自己的“处理队列”组件.
找到窃听器了吗? 特质请求? 放在这。 。 。 .
安装
创建一个 convex.config.ts 应用程序中的文件 convex/ 文件夹并安装
通过调用 use数字 :
// convex/convex.config.ts
import { defineApp } from "convex/server";
import batchWorker from "@convex-dev/batch-worker/convex.config.js";
const app = defineApp();
app.use(batchWorker, { env: { LOG_LEVEL: "REPORT" } });
export default app;
这个 env 选项是可选的,上面显示的默认值。 默认挂载名称为
batchWorker (即说. components.batchWorker) (中文(简体) ). 这个在另一个内部
组件: 调用 component.use(batchWorker) 在您的组件中
convex.config.ts。 。 。 .
使用量
在自己的表格中插入工作,然后调用 ping。提供返回的查询
下一个批次或 idle,并进行变异。 确保:
批量形状总是匹配查询和突变,使用
defineBatchWorkerValidators 以获取全部四个参数并返回验证符.
(单位:千美元)kind: "work" 可选的——返回 { batch } 独活也行. )
在批次的同时,查询返回 a 光标 说它有多远。 这个
组件用批量输入,然后按 args.cursor 下一个
时间,所以每次扫描 恢复上次扫描停止的地方。 见
为什么这很重要。 。 。 .
import { v } from "convex/values";
import { defineBatchWorkerValidators, ping } from "@convex-dev/batch-worker";
import { components, internal } from "./_generated/api";
import { internalMutation, internalQuery, mutation } from "./_generated/server";
const BATCH_SIZE = 10;
// Insert work, then make sure the loop is running.
export const addEvent = mutation({
args: { value: v.number() },
handler: async (ctx, { value }) => {
// `db.vars.commitTs` resolves at commit to an int64 ordered by commit
// order — that's what makes it safe to use as a cursor. See below.
await ctx.db.insert("events", { value, insertedAt: ctx.db.vars.commitTs });
await ping(ctx, components.batchWorker, {
name: "events", // distinct names give you independent queues
workQuery: internal.example.getBatch,
workerMutation: internal.example.processBatch,
});
},
});
const vEvents = v.array(v.object({ id: v.id("events"), value: v.number() }));
// One declaration gives the query's args (`{ name, cursor? }`) and returns,
// plus the mutation's args. `cursor` defaults to a commit timestamp.
const { vQueryArgs, vQueryReturns, vMutationArgs } =
defineBatchWorkerValidators({ batch: { events: vEvents } });
// Return the next batch of work, or `idle` when there's nothing to do.
export const getBatch = internalQuery({
args: vQueryArgs, // `name` lets one query serve many queues
returns: vQueryReturns,
handler: async (ctx, { cursor }) => {
// Resume where the last batch stopped.
const events = await ctx.db
.query("events")
.withIndex("insertedAt", (q) => q.gte("insertedAt", cursor ?? 0n))
.take(BATCH_SIZE);
if (events.length === 0) {
return { kind: "idle" as const };
// Or, if you know when the next item is due:
// return { kind: "idle" as const, timeoutMs: 30_000 };
}
return {
kind: "work" as const,
batch: { events: events.map((e) => ({ id: e._id, value: e.value })) },
// Rows come back in commit order, so the last one is how far we got.
cursor: events.at(-1)!.insertedAt,
};
},
});
// Process one batch. The worker owns cleanup — delete what you process!
export const processBatch = internalMutation({
args: vMutationArgs,
handler: async (ctx, { events }) => {
for (const { value, id } of events) {
// ... do the work (sum, call an API, schedule downstream jobs, etc.) ...
// Clean up the work you processed.
await ctx.db.delete("events", id);
}
// Returning nothing re-runs immediately to drain the rest.
},
});
构成部分 不清理你的工作给你 你的工人变异是 负责删除(或标记完成/推进过去)其行 已处理, 否则下一个查询将再次返回 .
在查询中获取工作, 在变异中处理
将工作查询和工人的变异视为 两个单独的交易 (在引擎盖下,他们运行 同一数据库的快照,但他们的行为 (以冲突为目的) 此分割可以同时处理 不在获取时添加数据库中的工作冲突 .
- 这个 工作查询 作为快照读取 无读
依赖关系。它可以扫描队列(
.take(BATCH_SIZE)超过指数) 与相并插入. - 这个 工人突变 是一个普通的交易: 它读取的每个范围都变为 a 读取依赖。 如果它重排队列中的工作范围, 则同时进行 插入会相冲突,所以范围只在工作查询中读取.
所以说: 在查询中获取批次, 并传递突变所需的所有信息
结束 batch,因此突变不需要再调整. 注:当
突变更新行 _id这取决于这些文件,所以它是
防止这些文件同时被更改.
// ✅ Query scans; mutation gets the data handed to it.
return {
kind: "work" as const,
batch: { events: rows.map((e) => ({ id: e._id, value: e.value })) },
};
动态地引导循环
你的工人可能会变异 { debounceMs } 以阻断回路,和a
cursor 以覆盖查询'(参见 光标:
return {
// Don't run again — and ignore pings — for at least this long (debounce).
debounceMs: 30_000,
};
(此返回值空格在循环运行时将批次删除; ping's
config.debounceMs 下面是被闲置唤醒时的延迟
插入可以累积成一批。 ) 弹出时的"失"是
无害:当弹出后,循环总是会重新运行您的工作查询,所以
同时插入的任何东西都被取出.
同样,如果没有工作,您的查询可以返回
{ kind: "idle", timeoutMs } 以确保它醒来 过了一段时间,即使
ping 没有叫。 一只平还立刻唤醒了它.
return {
kind: "idle",
// Keep polling this long before transitioning to idle.
cooldownMs: 10_000,
// How often to poll while cooling down.
pollIntervalMs: 250,
// After cooling down, wake again after at most this long even if no ping
// arrives. Measured from this query response, so re-run it each query if you
// want it to track a fixed deadline. A ping still wakes it sooner.
timeoutMs: 60_000,
};
Ing
平能保证工人运行.
选项 :
name:队名为平. 你可以给每个队列一个不同的name。此名称将传递到您的查询中args.name,所以一个查询/更改 配对可以服务许多队列. 平时只有一个工人有静电 名称.workQuery:返回下一批工作的查询,或idle。 。 。 .workerMutation: 处理一批的突变.config:工人的可选配置. 它储存在工人和 变化时刷新.
await ping(ctx, components.batchWorker, {
name: "events",
workQuery: internal.example.getBatch,
workerMutation: internal.example.processBatch,
config: {
// wait before the first batch so a burst accumulates
debounceMs: 100,
},
});
光标
组件为您存储光标: 无论您返回的工作查询
和一批货一起回来 args.cursor 下一个电话。 承诺
在与批次相同的交易中, 所以,如果你的工人突变抛出
光标留在原地,批次再次发出.
默认光标类型是
执行时间戳
(Convex ^ 1.43),它避免了必须处理违法犯罪. 添加一个
v.commitTs() 字段到表格中,工人扫描、索引和写入
ctx.db.vars.commitTs 在每个插入或补丁上插入:
// convex/schema.ts
myWork: defineTable({
updatedAt: v.commitTs(), // resolves at commit to an int64 in commit order
...fields
}).index("updatedAt", ["updatedAt"]),
await ctx.db.insert("myWork", { updatedAt: ctx.db.vars.commitTs, ... });
await ctx.db.patch("myWork", id, { updatedAt: ctx.db.vars.commitTs, ... });
执行 没有 使用情况 _creationTime。当一个变异 开始,这样一行
由慢突变插入可降落 后面 已经排好工了
问过去.
扫描为 gte。 。 。 . 单个突变插入的每分之一
因此,一个批次可以结束 在这样的领带中间,下一次扫描有
从领带开始接其他的 最多从领带费用中重新阅读
一个突变的价值 已经处理的行.
要结束每批标注在平分边界上,请在
将时间戳入批次并扫描 gt. 这一过程
工作同时添加了"一次",只要符合一次交易. 见
总计示例。 。 。 .
其他光标类型
光标可以是任何 Convex 值,例如 paginator 光标来源
convex-helpers,可以捕捉索引范围
[commitTs, _creationTime, _id] 所以,你很清楚 你离开了哪里。 过
cursor 改为 defineBatchWorkerValidators 所有四个验证人都拿起它:
const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
defineBatchWorkerValidators({
batch: { ids: v.array(v.id("myWork")) },
cursor: v.string(),
});
ping 从工作查询中取出光标类型 cursor 参数和检查
两种返回类型对它, 所以一个地方不匹配.
从工人突变中调整光标
工人的突变可以返回a cursor 超过查询。 用这个
当批量取得部分进展时,您可以找出它达到多大程度,例如.
使用批次所携带的每个项目的时间戳:
return { cursor: lastProcessed.updatedAt };
如果不返回,它将推迟到工作查询返回的光标, 或保留旧值。 注:如果光标不推进 而你没有更新任何读取的查询, 你可能会在无限的 循环.
读取并重置光标
工作查询收到光标的参数 。 在别处读,说从
查询计数待完成的工作,请如 status数字 :
const from = ((await ctx.runQuery(components.batchWorker.lib.getCursor, {
name: "events",
})) ?? 0n) as bigint;
lib.setCursor 覆盖它,然后清理它 cursor 被省略,所以
下一次扫描从前面开始。 用于迁移和恢复:循环
所以在工人忙碌的时候打个电话
可能发生冲突。 如果工人还不存在,就会被扔出去ping 创建
) (中文(简体) ).
为什么需要光标?
原因是Convex跟踪文档变化的技术细微. 在内部, Convex 在索引中为一些文件存储以前的版本 时间(~分钟). 旧的价值观不会在结果中归还给联署机构,但它可以 如果需要扫描和丢弃这些条目, 请慢下来查询 .
在上述简单的例子中,a .take(BATCH_SIZE) 从桌子前方
然后删除所有这些文件,结果导致
表格的索引将充满已删除的文档,然后在下一个查询中扫描
在进入下一批工作之前.
解决方案是保持光标以跟踪您的进度, 并做您的下一个 查询从光标开始,允许它跳过索引的区域 充满了以前版本或以前处理过的作品.
使用大小写: 更新非正态集合
为了不断更新非正常化的集合(如计数或总和),可以使用 批处理背景更新的批发商,避免写入冲突.
关于完整可操作的例子,见 总数.ts。 。 。 .
注:这一模式意味着汇总文件并非立即 反映变化,所以您在读取时需要稍有停滞的数据 这个值。 如果需要完整更新的值,您有几种选项:
- 开始立即更新值 重新审视懒惰的方法 如果/当它被证明是一个真正的缩放问题。 Convex 已重试 数据库多次相冲突,所以您可能可以使用简单的 办法.
- 阅读所有更新,并将其与 Stale 值动态结合. 何时 由询问完成,它将保持一致性和反应性。 下行:可能 需要读取大量更新.
- 使用类似 Sharded 计数器组件, 它可以平行写 并读取全部的硬盘. 这是立即一致的, 但需要阅读每一个硬,需要 用于手动调制以避免冲突.
使用大小写: 用动作连接工作
工人的突变是一种交易,所以不能自己调用外部API. 改为:
需要进行的批量工作 fetch呼叫LLM,击中第三方API..
工人变异 索赔 批次( 标记开始的行, 这样查询
然后安排行动 或排队
劳动池. 动作会起作用 并且需要突变来做出结果
回来.
完整工作代码在 利率限制( L). ts。 分批量的收费限额也包括在内.
提示: 如果您使用工作池, 您可以配置动作的重试, 并执行
中的结果或处理任何错误 onComplete 处理器,它有很好的
被叫过一次 你可以通过信息
批次编号 context 联 合 国 onComplete 处理器可以更新
即使是在失败的情况下。 这也将有助于您管理并行的同步
所以现场申请不会受到激增的工作量的影响.
不使用工作池的,可以发现预定工作的失败和 管理自己重试.
限制费率战略
您可以使用批量工人 限速器组件 改为 绑定循环运行的速度,例如遵守外部API限制.
完整工作代码在 利率限制( L). ts。 。 。 .
限制利率的 LLM 调用提示 :
- 您可以使用先前的 LLM 请求返回的符号获取准确性 前面信件的符号数.
- 在请求后达到利率限制对API来说是一个很好的权衡 有一些优雅 周围的请求, 而在那里你不知道 满 使用前置。 请求后调整有助于拖延今后的请求,因此 即使你暂时离开,你的总体使用量也符合标准 {\fn华文仿宋\fs16\1chd1d1d1}在它上面.
- 通过使用限速器 储备金 功能,可以避免试验
错误:保留从来不拒绝,而是保留未来的能力,如果
有必要。 它返回一个
retryAfter价值指示您等待多久 在执行你的请求之前. - 使用
retryAfter两种方式—— 安排你实际上的呼叫 {\fn黑体\fs22\bord1\shad0\3ahbe\4ah00\fscx67\fscy66\2chffffff\3ch808080}被允许来做这个debounceMs以延迟组装下一批. - 最简单的版本是使用
rateLimiter.check(ctx, name)改为 确定任何必要的延迟,以计入先前的请求,然后消耗 基于全部使用量的限制rateLimiter.limit(ctx, name, { count: totalTokens, reserve: true })之后 请求完成. - 为了更精确的跟踪, 您可以在请求之前保留容量
打开 投入 指数估计数,然后计入 产出 符号
之后,与
reserve: true从两个地方. - 如果你宁愿推迟工人的变异 直到有足够的容量
工作查询可以调用
rateLimiter.check(...)(只读)和返回{ kind: "idle", timeoutMs: retryAfter }换句话说.
处理失败
如果你的工作查询或工人的变异抛出,循环就会消失和活泼
显示器在~后重新启动monitorLagMs未处理的行仍然在
您的表格和光标一直停留在原地,所以查询会发出 不变
又来批次。这给你至少一次的处理, 但也意味着一个
总是扔出毒物的能挡住队列 可能失败的工作
中,在工人突变中捕捉出错,并在表格中隔离坏文件
为 aync 调试.
这是一个低等原始的, 相对像工作池或工作流, 所以,你必须处理 特殊情况自己.
停止恢复( R)
stop 完全停止处理: 循环停止 ping 被忽略,所以没有
新工作开始. start 恢复( 重用最后一个 ping编
查询/更改)。 召集他们:
await ctx.runMutation(components.batchWorker.lib.stop, { name: "events" });
// ...later, when you want it processing again:
await ctx.runMutation(components.batchWorker.lib.start, { name: "events" });
status 报告运行状态,包括工人是否 stopped。 。 。 .
ping 对 start
ping创建工人一叫,闲时恢复. 这个 也唤醒了在闲地上睡觉的循环timeoutMs铃声是无声的 当它已经运行时,下一个运行即将(在~1s范围内),在 adebounceMs窗口(工作在弹出后被取出),或 当工人被拦住的时候.start恢复一个stopped工人们。 小平不会再继续了stopped。 。 。 .
测试 convex-test
在测试设置中用导出自
@convex-dev/batch-worker/test数字 :
/// <reference types="vite/client" />
import { convexTest } from "convex-test";
import schema from "./schema";
import batchWorker from "@convex-dev/batch-worker/test";
const modules = import.meta.glob("./**/*.*s");
export function initConvexTest() {
const t = convexTest(schema, modules);
batchWorker.register(t); // pass a name if you mounted it under one
return t;
}
ping 本身立即返回——循环通过预定函数运行——所以
在您驱动调度器之前, 没有任何进程 。 运行回路( 和一切)
以假定时器完成:
vi.useFakeTimers();
await t.mutation(api.example.addEvent, { value: 5 });
await t.finishAllScheduledFunctions(vi.runAllTimers);
见 实例. test.ts 财务报告和审定财务报表 设置. test.ts。 。 。 .
注:使用 convex-test (^ 0.055-α.0)用于 v.commitTs() 支助.
参见示例应用中的完整工作示例: 实例.ts (基本队列), 总数.ts (非正常总量)和 利率限制( L). ts (同步LLM批次).
发展
用重建组件的文件监视器运行示例应用程序 :
npm i
npm run dev
运行 npm run dev:frontend (在第二个终端) 与它互动通过
vite 演示用户界面 example/src。它练习所有三个
工人——基本队列,活计分板,以及Aync LLM批次.
Convex本身通过
@convex-dev/static-hosting, (中文(简体) ),
发表于 https://<deployment>.convex.site。见
实例/README.md 用于部署命令.
如何运作
| 图表 | 作者: | 阅读者 |
|---|---|---|
workers | ping编号 :start编号 :loop (过渡) | ping编号 :start显示器 |
workerState | loop (每个迭代) | loop显示器 |
高转圈状态在 workerState (产生,心跳
计划跑者、显示器和光标,它们沿着一个补丁
循环已经进行每个迭代, 与很少写的分开 workers
doc(持有控件、配置和运行状态): idle 编号 : running 编号 :
stopped,加指针 workerState) (中文(简体) ). 这样 ping编号 :start —— 说
你们每插入一个字,就应当宣读 workers 和不相冲突地返回
(OCC) 与快扫圈. 单调 generation (单位:千美元) workerState(中文(简体) )
保证每次只运行一个循环链: 被取代的循环显示
生成和退出不匹配 。 现场监视器已经安排好了
~ (法语)monitorLagMs 之后 循环是下一次运行, 并按循环保持后推
运行中,所以它只使用火(并重新启动回路),如果回路实际死亡.