@ batchactions/ core (帮助)

核心批量处理引擎 @batchactions 生态系统.

当您需要低级控制批量管弦、状态过渡、事件和基础设施适配器时使用此软件包.

安装

npm install @batchactions/core

你得到的东西

• BatchEngine 用于流式批量执行
• 工作周期控制:开始、暂停、恢复、中止、处理块
• 任务、批量和记录进度的域事件
• 内建来源: BufferSource, (中文(简体) ), FilePathSource, (中文(简体) ), StreamSource, (中文(简体) ), UrlSource
• 已建成的国营商店: InMemoryStateStore, (中文(简体) ), FileStateStore
• 分配合同: DistributedStateStore, (中文(简体) ), BatchReservation,帮助者

快速启动

import { BatchEngine, BufferSource, InMemoryStateStore } from '@batchactions/core';

const engine = new BatchEngine({
  stateStore: new InMemoryStateStore(),
  batchSize: 100,
  maxConcurrentBatches: 2,
  continueOnError: true,
});

engine.from(new BufferSource(JSON.stringify([{ id: 1 }, { id: 2 }])), {
  async *parse(chunk) {
    const rows = JSON.parse(chunk.toString()) as Array<Record<string, unknown>>;
    for (const row of rows) {
      yield row;
    }
  },
});

await engine.start(async (record) => {
  await processRecord(record);
});

记忆记录

当您的数据已经存在内存时( 数据库结果, API 响应等), 跳过源/ parser 管道 :

import { BatchEngine } from '@batchactions/core';

const accounts = [
  { id: 'acc-1', channel: 'email', status: 'active' },
  { id: 'acc-2', channel: 'sms', status: 'active' },
];

const engine = new BatchEngine({
  batchSize: 50,
  maxConcurrentBatches: 4,
  continueOnError: true,
  maxRetries: 2,
});

engine.fromRecords(accounts);

await engine.start(async (record) => {
  await healthCheck(record);
});

此外,还利用同步发电机进行懒惰评价:

async function* fetchPages(): AsyncIterable<Record<string, unknown>> {
  let page = 1;
  while (true) {
    const results = await api.getPage(page++);
    if (results.length === 0) break;
    for (const item of results) yield item;
  }
}

engine.fromRecords(fetchPages());

无服务器 / 断开处理

使用 processChunk() 与 restore() 以在多个引用中渐进处理记录——对有执行时限的无服务器环境(AWS Lambda、Vercel函数、Cloudflare Workers)的理想.

基本图案

import { BatchEngine, InMemoryStateStore } from '@batchactions/core';

const stateStore = new InMemoryStateStore(); // or FileStateStore, SequelizeStateStore, etc.

export async function handler(event: { jobId?: string }) {
  let engine: BatchEngine;

  if (event.jobId) {
    // Subsequent invocation: restore from persisted state
    const restored = await BatchEngine.restore(event.jobId, { stateStore, batchSize: 100 });
    if (!restored) throw new Error('Job not found');
    engine = restored;
  } else {
    // First invocation: create a new engine
    engine = new BatchEngine({ stateStore, batchSize: 100, continueOnError: true });
  }

  // The source must be re-fed on every invocation (it will be re-streamed)
  engine.from(source, parser);

  const result = await engine.processChunk(
    async (record) => { await processRecord(record); },
    { maxDurationMs: 25_000 },  // stop before Lambda's 30s timeout
  );

  if (!result.done) {
    // Schedule the next invocation with the job ID
    await scheduleNext({ jobId: result.jobId });
  }

  return { done: result.done, processed: result.totalProcessed };
}

maxRecords 对 maxDurationMs

两个选项可以合并. 块停止当 无论是 已达到极限.

重要说明

• 重组要求: 来源必须在每次援引时通过 from() 或者说 fromRecords()。引擎从头进行再流,并自动跳过已完成的批次.
• 批量级别边界:在批次之间,而不是记录之间检查春克限制. 一批总是在块停止前完成. 用一个较小的 batchSize 为了更好的控制.
• 使用持久状态 Store数字 : InMemoryStateStore 在进程重新启动之间失去状态。 用于生产无服务器,配对 @batchactions/state-sequelize 或者说 @batchactions/state-prisma。 。 。 .
• done 旗帜编号: ChunkResult.done 财产可靠地表明是否已处理所有记录。 进展衡量尺度totalRecords, (中文(简体) ), percentage在来源被完全消耗之前,可能显示部分总数.

主要出口

• BatchEngine
• BatchSplitter
• EventBus, (中文(简体) ), JobContext
• JobStatus, (中文(简体) ), BatchStatus
• BufferSource, (中文(简体) ), FilePathSource, (中文(简体) ), StreamSource, (中文(简体) ), UrlSource
• InMemoryStateStore, (中文(简体) ), FileStateStore
• isDistributedStateStore

完整类型出口,见 packages/core/src/index.ts。 。 。 .

兼容性

• 节点.js >= 20.0.0

相关软件包

• @batchactions/import: 高级进口外观
• @batchactions/distributed: 多工管弦乐
• @batchactions/state-sequelize: SQL 持久性适配器(Sequelize)
• @batchactions/state-prisma: SQL 持久性适配器 (Prisma v6/v7)

链接

• 仓库 : https://github.com/vgpastor/batchactions/tree/main/packages/core 页面存档备份,存于互联网档案馆
• 问题: https://github.com/vgpastor/batchactions/ issues (中文(简体) )
• 贡献指南 : https://github.com/vgpastor/batchactions/blob/main/CONTRIPIBUTING.md 互联网档案馆的存檔,存档日期2014-12-22

许可证

麻省理工学院