iii-queue-processing

队列处理

可与以下工具比较：BullMQ、Celery、SQS

核心概念

在适合任务的情况下使用以下概念。并非每种队列配置都需要全部概念。

• 命名队列在 iii-config.yaml 的 queue_configs 下声明
• 标准队列并发处理作业；FIFO 队列则保持处理顺序
• TriggerAction.Enqueue({ queue }) 将作业分发到命名队列
• 失败的作业会通过指数退避机制自动重试，最多达到 max_retries
• 重试次数耗尽的作业会进入死信队列，以供检查
• 每个消费者函数都会接收作业负载和 messageReceiptId

架构

生产者函数
  → TriggerAction.Enqueue({ queue: 'task-queue' })
    → 命名队列（标准或 FIFO）
      → 消费者 registerFunction 处理程序
        → 成功 / 通过退避机制重试
          → 死信队列（达到 max_retries 后）

使用的 iii 基元

基元	用途
registerFunction	定义处理作业的消费者
trigger({ ..., action: TriggerAction.Enqueue({ queue }) })	将作业分发到命名队列
messageReceiptId	确认或跟踪单个作业的处理情况
iii-config.yaml 中的 queue_configs	声明队列的并发数和重试设置

参考实现

完整的可运行示例见 ../references/queue-processing.js：生产者将作业加入队列，消费者则通过重试逻辑处理作业。

另有 Python 版本：../references/queue-processing.py

另有 Rust 版本：../references/queue-processing.rs

常见模式

使用此模式的代码通常会在适用时包含：

• registerWorker(url, { workerName }) — 初始化工作进程
• registerFunction(id, handler) — 定义消费者
• trigger({ function_id, payload, action: TriggerAction.Enqueue({ queue }) }) — 将作业加入队列
• payload.messageReceiptId — 跟踪或确认作业
• trigger({ function_id: 'state::set', payload }) — 处理后持久化结果
• const logger = new Logger() — 记录每个作业的结构化日志

调整此模式

在适用于任务时，使用以下调整方式。

• 如果作业顺序很重要（例如顺序执行流水线步骤），请选择 FIFO 队列
• 在队列配置中设置 max_retries 和 concurrency，以适配工作负载
• 对于多阶段流水线，可串联多个队列（队列 A 的消费者将作业加入队列 B）
• 为实现幂等性，在处理前检查状态，避免重试时重复执行

引擎配置

在 iii-config.yaml 的 queue_configs 下声明命名队列，并为每个队列设置 max_retries、concurrency、type 和 backoff_ms。完整的带注释配置参考见 ../references/iii-config.yaml。

模式边界

• 如果任务只需要即发即弃，不需要重试或排序，请优先使用 iii-trigger-actions 和 TriggerAction.Void()。
• 如果失败的作业需要特殊处理或告警，请优先使用 iii-dead-letter-queues 实现死信队列消费者。
• 如果任务需要逐步编排并包含分支，请优先使用 iii-workflow-orchestration。
• 如果主要需求是可靠地异步执行作业并支持重试，请继续使用 iii-queue-processing。

适用场景

• 当任务主要涉及 iii 引擎中的 iii-queue-processing 时，使用此技能。
• 当请求直接要求此模式或等效实现时，使用此技能。

边界

• 切勿将此技能用作无关任务的通用回退方案。
• 如果有更具体的 iii 技能更适合，就不得使用此技能。
• 应用此技能中的示例前，务必先确认环境和安全约束。