# @convex-dev/batch-worker 中文文档

规范地址：https://npmcn.com/package/%40convex-dev/batch-worker/

版本：0\.3\.6
npm 元数据更新时间：2026-10-10T01:01:42.446Z
中文译文更新时间：2026-11-10T17:33:55.742Z
软件包许可：Apache-2\.0

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40convex-dev/batch-worker
原文对照：https://npmcn.com/package/%40convex-dev/batch-worker/?lang=original
项目主页：https://github.com/get-convex/batch-worker#readme
源码仓库：https://github.com/get-convex/batch-worker

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：276 kB，77 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- 运行依赖：1 个
- peer 依赖：convex ^1\.43\.0
- 首次发布：2026-06-18T18:16:33.399Z
- 最新版本发布：2026-10-10T01:01:42.252Z
- 近 30 天下载量：3929145（2026-10-11 至 2026-11-09，npm 官方下载统计）
- 已发布版本数：15（近 12 个月 15 个）

## 中文简介

Convex的批量工人组件.

## 安装

```sh
npm install @convex-dev/batch-worker
```

也可以使用：`pnpm add @convex-dev/batch-worker`、`yarn add @convex-dev/batch-worker`、`bun add @convex-dev/batch-worker`

## README

# 批发工人

[![npm version](https://badge.fury.io/js/@convex-dev%2Fbatch-worker.svg)](https://badge.fury.io/js/@convex-dev/batch-worker) [![Convex Component](https://www.convex.dev/components/badge/batch-worker)](https://www.convex.dev/components/batch-worker)

通过定义循环工人在自己的表格中加工批次的工作 自动的, 与排程, 解跳, 和恢复构建在.

你带来两个功能:

-   A 类 **工作查询** 返回下一批工作,或明确闲置.
-   A 类 **工人突变** 处理批量.

插入工作后,请拨 `ping(...)`. 本部分负责其余部分:

-   每个工人一次运行一个循环.
-   支持拆弹,所以它们一起批量.
-   保持"温暖"的循环 短时间的投票冷却 所以一丝一毫的新工作 不打运行状态.
-   使用快照读取时会排出 这样同时插入不会引起OCC 重复一遍,在闲逛之前用真实的读数确认,这样就不会失去任何东西.
-   当队列排出后闲置, 下次自动重新启动 p.
-   监视循环和 **如果它死了就重启它** (例如,意外事件) 错误),记录失败,以便您可以对此提醒.

这是组件背后的图案,比如 [工作组合](https://github.com/get-convex/workpool),提取后可以构建 。您自己的“处理队列”组件.

找到窃听器了吗? 特质请求? [放在这](https://github.com/get-convex/batch-worker/issues)。 。 。 .

## 安装

创建一个 `convex.config.ts` 应用程序中的文件 `convex/` 文件夹并安装 通过调用 `use`数字 :

```ts
// 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` 下一个 时间,所以每次扫描 恢复上次扫描停止的地方。 见 [为什么这很重要](https://npmcn.com/package/%40convex-dev/batch-worker/#readme-why-is-a-cursor-necessary)。 。 。 .

```ts
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`这取决于这些文件,所以它是 防止这些文件同时被更改.

```ts
// ✅ 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` 以覆盖查询'(参见 [光标](https://npmcn.com/package/%40convex-dev/batch-worker/#readme-the-cursor):

```ts
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` 没有叫。 一只平还立刻唤醒了它.

```ts
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`:工人的可选配置. 它储存在工人和 变化时刷新.

```ts
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` 下一个电话。 承诺 在与批次相同的交易中, 所以,如果你的工人突变抛出 光标留在原地,批次再次发出.

默认光标类型是 [执行时间戳](https://docs.convex.dev/database/advanced/commit-timestamp) (Convex ^ 1.43),它避免了必须处理违法犯罪. 添加一个 `v.commitTs()` 字段到表格中,工人扫描、索引和写入 `ctx.db.vars.commitTs` 在每个插入或补丁上插入:

```ts
// convex/schema.ts
myWork: defineTable({
  updatedAt: v.commitTs(), // resolves at commit to an int64 in commit order
  ...fields
}).index("updatedAt", ["updatedAt"]),
```

```ts
await ctx.db.insert("myWork", { updatedAt: ctx.db.vars.commitTs, ... });
await ctx.db.patch("myWork", id, { updatedAt: ctx.db.vars.commitTs, ... });
```

执行 **没有** 使用情况 `_creationTime`。当一个变异 _开始_,这样一行 由慢突变插入可降落 _后面_ 已经排好工了 问过去.

**扫描为 `gte`。 。 。 .** 单个突变插入的每分之一 因此,一个批次可以结束 在这样的领带中间,下一次扫描有 从领带开始接其他的 最多从领带费用中重新阅读 一个突变的价值 已经处理的行.

要结束每批标注在平分边界上,请在 将时间戳入批次并扫描 `gt`. 这一过程 工作同时添加了"一次",只要符合一次交易. 见 [总计示例](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/aggregates.ts)。 。 。 .

#### 其他光标类型

光标可以是任何 Convex 值,例如 `paginator` 光标来源 `convex-helpers`,可以捕捉索引范围 `[commitTs, _creationTime, _id]` 所以,你很清楚 你离开了哪里。 过 `cursor` 改为 `defineBatchWorkerValidators` 所有四个验证人都拿起它:

```ts
const { vQueryArgs, vQueryReturns, vMutationArgs, vMutationReturns } =
  defineBatchWorkerValidators({
    batch: { ids: v.array(v.id("myWork")) },
    cursor: v.string(),
  });
```

`ping` 从工作查询中取出光标类型 `cursor` 参数和检查 两种返回类型对它, 所以一个地方不匹配.

#### 从工人突变中调整光标

工人的突变可以返回a `cursor` 超过查询。 用这个 当批量取得部分进展时,您可以找出它达到多大程度,例如. 使用批次所携带的每个项目的时间戳:

```ts
return { cursor: lastProcessed.updatedAt };
```

如果不返回,它将推迟到工作查询返回的光标, 或保留旧值。 注:如果光标不推进 而你没有更新任何读取的查询, 你可能会在无限的 循环.

#### 读取并重置光标

工作查询收到光标的参数 。 在别处读,说从 查询计数待完成的工作,请如 `status`数字 :

```ts
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](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/aggregates.ts)。 。 。 .

注:这一模式意味着汇总文件并非立即 反映变化,所以您在读取时需要稍有停滞的数据 这个值。 如果需要完整更新的值,您有几种选项:

1.  开始立即更新值 重新审视懒惰的方法 如果/当它被证明是一个真正的缩放问题。 Convex 已重试 数据库多次相冲突,所以您可能可以使用简单的 办法.
2.  阅读所有更新,并将其与 Stale 值动态结合. 何时 由询问完成,它将保持一致性和反应性。 下行:可能 需要读取大量更新.
3.  使用类似 Sharded 计数器组件, 它可以平行写 并读取全部的硬盘. 这是立即一致的, 但需要阅读每一个硬,需要 用于手动调制以避免冲突.

### 使用大小写: 用动作连接工作

工人的突变是一种交易,所以不能自己调用外部API. 改为: 需要进行的批量工作 `fetch`呼叫LLM,击中第三方API.. 工人变异 _索赔_ 批次( 标记开始的行, 这样查询 然后安排行动 或排队 劳动池. 动作会起作用 并且需要突变来做出结果 回来.

完整工作代码在 [利率限制( L). ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/rateLimited.ts)。 分批量的收费限额也包括在内.

提示: 如果您使用工作池, 您可以配置动作的重试, 并执行 中的结果或处理任何错误 `onComplete` 处理器,它有很好的 被叫过一次 你可以通过信息 批次编号 `context` 联 合 国 `onComplete` 处理器可以更新 即使是在失败的情况下。 这也将有助于您管理并行的同步 所以现场申请不会受到激增的工作量的影响.

不使用工作池的,可以发现预定工作的失败和 管理自己重试.

### 限制费率战略

您可以使用批量工人 [限速器组件](https://www.convex.dev/components/rate-limiter) 改为 绑定循环运行的速度,例如遵守外部API限制.

完整工作代码在 [利率限制( L). ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/rateLimited.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`编 查询/更改)。 召集他们:

```ts
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范围内),在 a `debounceMs` 窗口(工作在弹出后被取出),或 当工人被拦住的时候.
-   **`start`** 恢复一个 `stopped` 工人们。 小平不会再继续了 `stopped`。 。 。 .

### 测试 `convex-test`

在测试设置中用导出自 `@convex-dev/batch-worker/test`数字 :

```ts
/// <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` 本身立即返回——循环通过预定函数运行——所以 在您驱动调度器之前, 没有任何进程 。 运行回路( 和一切) 以假定时器完成:

```ts
vi.useFakeTimers();
await t.mutation(api.example.addEvent, { value: 5 });
await t.finishAllScheduledFunctions(vi.runAllTimers);
```

见 [实例. test.ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/example.test.ts) 财务报告和审定财务报表 [设置. test.ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/setup.test.ts)。 。 。 .

注:使用 `convex-test` (^ 0.055-α.0)用于 `v.commitTs()` 支助.

参见示例应用中的完整工作示例: [实例.ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/example.ts) (基本队列), [总数.ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/aggregates.ts) (非正常总量)和 [利率限制( L). ts](https://npmcn.com/package/%40convex-dev/batch-worker/example/convex/rateLimited.ts) (同步LLM批次).

## 发展

用重建组件的文件监视器运行示例应用程序 :

```sh
npm i
npm run dev
```

运行 `npm run dev:frontend` (在第二个终端) 与它互动通过 vite 演示用户界面 [`example/src`](https://npmcn.com/package/%40convex-dev/batch-worker/example/src)。它练习所有三个 工人——基本队列,活计分板,以及Aync LLM批次.

Convex本身通过 [`@convex-dev/static-hosting`](https://github.com/get-convex/static-hosting), (中文(简体) ), 发表于 `https://<deployment>.convex.site`。见 [实例/README.md](https://npmcn.com/package/%40convex-dev/batch-worker/example/README.md#deploy) 用于部署命令.

### 如何运作

| 图表 | 作者: | 阅读者 |
| --- | --- | --- |
| `workers` | `ping`编号 :`start`编号 :`loop` (过渡) | `ping`编号 :`start`显示器 |
| `workerState` | `loop` (每个迭代) | `loop`显示器 |

高转圈状态在 `workerState` (产生,心跳 计划跑者、显示器和光标,它们沿着一个补丁 循环已经进行每个迭代, 与很少写的分开 `workers` doc(持有控件、配置和运行状态): `idle` 编号 : `running` 编号 : `stopped`,加指针 `workerState`) (中文(简体) ). 这样 `ping`编号 :`start` —— 说 你们每插入一个字,就应当宣读 `workers` 和不相冲突地返回 (OCC) 与快扫圈. 单调 `generation` (单位:千美元) `workerState`(中文(简体) ) 保证每次只运行一个循环链: 被取代的循环显示 生成和退出不匹配 。 现场监视器已经安排好了 ~ (法语)`monitorLagMs` _之后_ 循环是下一次运行, 并按循环保持后推 运行中,所以它只使用火(并重新启动回路),如果回路实际死亡.

## 常见问题

### @convex-dev/batch-worker 是什么？

Convex的批量工人组件\.

### 如何安装 @convex-dev/batch-worker？

在项目目录运行 npm install @convex-dev/batch-worker，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：convex（^1\.43\.0）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @convex-dev/batch-worker 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @convex-dev/batch-worker 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### @convex-dev/batch-worker 可以商用吗？

@convex-dev/batch-worker 使用 Apache-2\.0 许可。宽松许可：允许商业使用、修改和再分发，并附带明确的专利授权；分发时需保留许可与 NOTICE 文件，修改过的文件要注明改动。以上为概括，不构成法律意见，请以许可原文为准。

### @convex-dev/batch-worker 的下载量有多少？

根据 npm 官方下载统计，@convex-dev/batch-worker 在 2026/10/11 – 2026/11/09 这 30 天内被下载 3,929,145 次，日均约 130,972 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @convex-dev/batch-worker 还在维护吗？

最新版本 0\.3\.6 发布于 2026/10/10（1 个月前）；至今共发布 15 个版本，其中过去 12 个月发布了 15 个；首次发布于 2026/06/19。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。