inngest-flow-control

Inngest 流量控制

掌握 Inngest 流量控制机制，以管理资源、防止系统过载并确保应用可靠性。本技能涵盖所有流量控制选项，并针对每种选项提供何时及如何使用的明确指导。

这些技能面向 TypeScript。 对于 Python 或 Go，请参阅 Inngest 文档，获取特定语言的指导。核心概念适用于所有语言。

快速决策指南

• “限制同时运行的数量” → 并发控制
• “将运行分散到一段时间内” → 节流
• “在某个时间段内达到 N 次后阻止运行” → 速率限制
• “等待活动停止，然后运行一次” → 防抖
• “此键一次只能运行一个任务” → 单例
• “分组处理事件” → 批处理
• “某些运行更重要” → 优先级

并发控制

适用场景： 限制正在执行的步骤数量（而非函数运行次数），以管理计算资源并防止系统过载。

关键要点： 并发限制控制的是活动代码执行，而不是函数运行。等待 step.sleep() 或 step.waitForEvent() 的函数不会计入限制。

基本并发控制

inngest.createFunction(
  {
    id: "process-images",
    concurrency: 5,
    triggers: [{ event: "media/image.uploaded" }]
  },
  async ({ event, step }) => {
    // Only 5 steps can execute simultaneously
    await step.run("resize", () => resizeImage(event.data.imageUrl));
  }
);

使用键进行并发控制（多租户）

使用 key 参数，按键的每个唯一值分别应用限制。

inngest.createFunction(
  {
    id: "user-sync",
    concurrency: [
      {
        key: "event.data.user_id",
        limit: 1
      }
    ],
    triggers: [{ event: "user/profile.updated" }]
  },
  async ({ event, step }) => {
    // Only 1 step per user can execute at once
    // Prevents race conditions in user-specific operations
  }
);

账户级共享限制

inngest.createFunction(
  {
    id: "ai-summary",
    concurrency: [
      {
        scope: "account",
        key: `"openai"`,
        limit: 60
      }
    ],
    triggers: [{ event: "ai/summary.requested" }]
  },
  async ({ event, step }) => {
    // Share 60 concurrent OpenAI calls across all functions
  }
);

各自的适用场景：

• 基本：保护数据库或限制整体容量
• 按键：确保多租户公平，防止“吵闹的邻居”问题
• 账户级：在多个函数之间共享配额（API 限制）

节流

适用场景： 控制函数随时间启动的速率，以应对 API 速率限制或平滑流量峰值。

与并发控制的主要区别： 节流限制函数运行的启动次数；并发控制限制步骤执行。

inngest.createFunction(
  {
    id: "sync-crm-data",
    throttle: {
      limit: 10, // 10 function starts
      period: "60s", // per minute
      burst: 5, // plus 5 immediate bursts
      key: "event.data.customer_id" // per customer
    },
    triggers: [{ event: "crm/contact.updated" }]
  },
  async ({ event, step }) => {
    // Respects CRM API rate limits: 10 calls/min per customer
    await step.run("sync", () => crmApi.updateContact(event.data));
  }
);

配置项：

• limit：每个时间段内可启动的函数数
• period：时间窗口（1s 至 7d）
• burst：允许额外立即启动的次数
• key：按键的每个唯一值应用限制

速率限制

适用场景： 设置硬性限制，以防止滥用或跳过过多重复事件。

与节流的主要区别： 速率限制会丢弃事件；节流会延迟事件。

inngest.createFunction(
  {
    id: "webhook-processor",
    rateLimit: {
      limit: 1,
      period: "4h",
      key: "event.data.webhook_id"
    },
    triggers: [{ event: "webhook/data.received" }]
  },
  async ({ event, step }) => {
    // Process each webhook only once per 4 hours
    // Prevents duplicate webhook spam
  }
);

适用情形：

• 防止 Webhook 重复
• 限制每位用户执行高成本操作的次数
• 防止滥用

防抖

适用场景： 等待一系列事件停止到达后，再处理最新事件。

inngest.createFunction(
  {
    id: "save-document",
    debounce: {
      period: "5m", // Wait 5min after last edit
      key: "event.data.document_id",
      timeout: "30m" // Force save after 30min max
    },
    triggers: [{ event: "document/content.changed" }]
  },
  async ({ event, step }) => {
    // Saves document only after user stops editing
    // Uses the LAST event received
    await step.run("save", () => saveDocument(event.data));
  }
);

特别适用于：

• 快速变化的用户输入（搜索、文档编辑）
• 嘈杂的 Webhook 事件
• 确保处理最新数据

优先级

适用场景： 根据动态数据，让某些函数运行先于其他运行。

inngest.createFunction(
  {
    id: "process-order",
    priority: {
      // VIP users get priority up to 120 seconds ahead
      run: "event.data.user_tier == 'vip' ? 120 : 0"
    },
    triggers: [{ event: "order/placed" }]
  },
  async ({ event, step }) => {
    // VIP orders jump ahead in the queue
  }
);

高级示例：

inngest.createFunction(
  {
    id: "support-ticket",
    priority: {
      run: `
        event.data.severity == 'critical' ? 300 :
        event.data.severity == 'high' ? 120 :
        event.data.user_plan == 'enterprise' ? 60 : 0
      `
    },
    triggers: [{ event: "support/ticket.created" }]
  },
  async ({ event, step }) => {
    // Critical tickets get highest priority (300s ahead)
    // High severity: 120s ahead
    // Enterprise users: 60s ahead
    // Everyone else: normal priority
  }
);

单例

适用场景： 确保同一时间只有一个函数实例运行。

跳过模式（保留当前运行）

inngest.createFunction(
  {
    id: "data-backup",
    singleton: {
      key: "event.data.database_id",
      mode: "skip"
    },
    triggers: [{ event: "backup/requested" }]
  },
  async ({ event, step }) => {
    // Skip new backups if one is already running for this database
    await step.run("backup", () => performBackup(event.data.database_id));
  }
);

取消模式（使用最新事件）

inngest.createFunction(
  {
    id: "realtime-sync",
    singleton: {
      key: "event.data.user_id",
      mode: "cancel"
    },
    triggers: [{ event: "user/data.changed" }]
  },
  async ({ event, step }) => {
    // Cancel previous sync and start with latest data
    await step.run("sync", () => syncUserData(event.data));
  }
);

批处理

适用场景： 将多个事件一起处理，以提高效率。

inngest.createFunction(
  {
    id: "bulk-email-send",
    batchEvents: {
      maxSize: 100, // Up to 100 events
      timeout: "30s", // Or 30 seconds, whichever first
      // `key` groups events into separate batches per unique value
      // This is different from expressions `if` which filters events
      key: "event.data.campaign_id" // Batch per campaign
    },
    triggers: [{ event: "email/send.queued" }]
  },
  async ({ events, step }) => {
    // Process array of events together
    const emails = events.map((evt) => ({
      to: evt.data.email,
      subject: evt.data.subject,
      body: evt.data.body
    }));

    await step.run("send-batch", () => emailService.sendBulk(emails));
  }
);

组合流量控制机制

示例：公平的 AI 处理

inngest.createFunction(
  {
    id: "ai-image-processing",
    // Global throttling for API limits
    throttle: {
      limit: 50,
      period: "60s",
      key: `"gpu-cluster"`
    },
    // Per-user concurrency for fairness
    concurrency: [
      {
        key: "event.data.user_id",
        limit: 3
      }
    ],
    // VIP users get priority
    priority: {
      run: "event.data.plan == 'pro' ? 60 : 0"
    },
    triggers: [{ event: "ai/image.generate" }]
  },
  async ({ event, step }) => {
    // Combines multiple flow controls for optimal resource usage
  }
);

专业提示： 大多数生产环境中的函数都能通过组合 1-3 种流量控制机制，获得最佳的可靠性和性能。