PAYATHON 2026

如何优化基于 SQS queue 的事件驱动 worker node 架构？

支付阿杰

结论

把可能耗时超过 15 分钟的付款处理从同步 HTTP 调用改为异步任务，整体方向合理，也能避开 ALB 的请求超时限制。

不过，不建议把 integrator API 的后处理逻辑和 invoice API 调用直接移入 payment API。否则，payment API 会同时负责付款处理、流程编排和开票，服务间的耦合也会随之增加。

建议采用下面的结构：

Client
  │
  ▼
Integrator API
  │ 1. 保存原始 payment details
  │ 2. 创建 operation/job
  │ 3. 投递 PaymentRequested
  │ 4. 返回 202 Accepted + operationId
  ▼
Payment Request Queue
  │
  ▼
Payment Worker
  │ 调用 legacy DBMS
  │
  ├──成功──► PaymentCompleted Queue
  │
  └──失败──► retry / DLQ
                │
                ▼
Integrator Worker / Post-processing Worker
  │ 执行原 integrator 的 post processing
  │
  ▼
Invoice Request Queue
  │
  ▼
Invoice Worker / Invoice API

这里的 “Integrator Worker” 不一定要做成新的微服务，也可以是 integrator API 的另一个 ECS worker 进程。重点是让 payment API 专注于付款领域，不要让它在不知不觉中变成整个业务流程的编排器。

1. 请求入口应立即返回任务标识

用户不需要立刻看到 invoice details，因此入口 API 没必要等待付款和开票全部完成。

建议返回：

HTTP/1.1 202 Accepted
Location: /operations/pay_01JABC...
Content-Type: application/json

{
  "operationId": "pay_01JABC...",
  "status": "PENDING"
}

客户端可以通过以下方式获取最终状态：

  • 轮询 GET /operations/{operationId}
  • WebSocket 或 Server-Sent Events
  • Webhook
  • 邮件或站内通知

状态可以设计为：

PENDING
PAYMENT_PROCESSING
PAYMENT_COMPLETED
POST_PROCESSING
INVOICE_PENDING
COMPLETED
FAILED

状态记录最好单独保存在 operation/job 表中，不要只依赖暂时存放在 SQS 里的消息。

2. SQS 只传引用，不传大对象

大型 payment details 应存放在外部存储中，SQS 消息只携带引用。这是常见的 claim-check pattern。

消息可以保持得很小：

{
  "eventType": "PaymentRequested",
  "eventVersion": 1,
  "operationId": "pay_01JABC",
  "paymentId": "payment-12345",
  "payload": {
    "store": "s3",
    "key": "payments/pay_01JABC/request.json",
    "checksum": "sha256:..."
  },
  "correlationId": "req-789",
  "createdAt": "2026-09-01T08:30:00Z"
}

SQS 的最大消息大小应以当前 AWS 文档和目标区域的实际限制为准。即使某个请求没有超过当前限制，也不建议把接近上限的大对象直接放进消息，因为消息越大，网络开销、重试成本和费用通常也越高。

外部存储可以按数据的使用方式选择：

  • 如果原始数据本质上是 JSON 文档或二进制对象，优先考虑 S3。
  • 如果处理过程中需要按字段查询、修改或建立索引，DocumentDB 更合适。
  • 如果已经使用 DocumentDB，而且数据模型确实需要数据库能力，可以继续使用。

如果只是暂存大型 payload，S3 通常更简单。它的成本较低，对象大小限制宽松，还可以通过 lifecycle rule 自动清理。

3. 不要只靠 worker 主动删除暂存数据

“成功后由 payment API 删除数据”只能处理正常路径，不能作为唯一的清理机制。以下情况都可能留下残余数据：

  • worker 处理成功，但在删除前崩溃；
  • 消息进入 DLQ；
  • legacy DBMS 长时间无响应；
  • 后处理或开票永久失败；
  • 消息被重复消费；
  • 部署或网络故障导致状态不一致。

还应配置兜底清理策略：

  • S3 使用 lifecycle expiration；
  • DocumentDB 使用 TTL index；
  • operation 记录保存 expiresAt；
  • 定期清理已经进入终态并超过保留期的数据；
  • 为 FAILED 状态和 DLQ 中的任务保留足够时间，以便排查和重放。

付款处理结束后，不要立即删除原始 payload。如果后处理、开票或人工重放还需要这些数据，应在整个 operation 到达终态后再清理，或者按照明确的审计周期保留。

4. Payment Worker 必须按照”至少一次投递”设计

SQS Standard queue 可能重复投递消息。即使使用 FIFO queue，也不能完全依靠队列去重来保证业务正确性。

Payment Worker 应实现业务层幂等：

INSERT INTO processed_operations (
    operation_id,
    step_name,
    status
)
VALUES (
    :operation_id,
    'PROCESS_PAYMENT',
    'STARTED'
)
ON CONFLICT (operation_id, step_name) DO NOTHING;

具体写法需要根据所用数据库调整，但每个处理步骤都应有唯一键，例如：

(operationId, stepName)

调用 legacy DBMS 时，最好携带稳定的业务幂等键：

idempotencyKey = operationId + ":payment"

如果 legacy DBMS 不支持幂等键，就需要在本地持久化调用状态和业务结果，再通过付款编号、交易编号等唯一约束防止重复扣款。支付场景不能只靠删除 SQS 消息来保证”只执行一次”。

5. 正确处理 visibility timeout

worker 收到消息后，在 visibility timeout 到期前，其他消费者看不到这条消息。如果 legacy DBMS 的处理时间可能达到 15 分钟，visibility timeout 必须覆盖这段时间，或者由 worker 定期续期。

伪代码如下：

message = sqs.receive_message(
    QueueUrl=queue_url,
    WaitTimeSeconds=20,
    VisibilityTimeout=300
)

receipt_handle = message["ReceiptHandle"]

try:
    while payment_is_running():
        result = poll_or_process_payment()

        sqs.change_message_visibility(
            QueueUrl=queue_url,
            ReceiptHandle=receipt_handle,
            VisibilityTimeout=300
        )

    save_result(result)
    publish_payment_completed(result)

    sqs.delete_message(
        QueueUrl=queue_url,
        ReceiptHandle=receipt_handle
    )
except RetryableError:
    # 不删除消息，让 visibility timeout 到期后重新投递
    raise

实际运行时还要注意：

  • visibility timeout 需要留出余量；
  • 必须在当前 timeout 到期前续期；
  • worker 关闭时应停止接收新消息，并为正在处理的任务预留 drain 时间；
  • ECS 部署和 Auto Scaling 缩容时需要考虑长任务；
  • visibility timeout 不是数据库调用本身的超时，legacy DBMS 调用仍然需要明确的客户端超时和取消策略。

如果一次数据库调用会阻塞 15 分钟，而且无法查询进度，worker 只能维持长连接等待。如果 legacy 系统支持提交任务后查询状态，最好将流程拆成”提交任务”和”检查结果”两个阶段，避免长时间占用 ECS worker。

6. 发布成功事件与删除原消息之间存在一致性问题

下面三个操作不是原子的：

保存付款结果
发布 PaymentCompleted
删除 PaymentRequested

任何两个步骤之间发生崩溃，都可能造成事件重复或状态缺失。例如，系统成功发布 PaymentCompleted 后，在删除原消息前崩溃，原消息就会再次被消费。

常见的处理方式有两种：

  1. 所有消费者都实现幂等，并允许事件重复。
  2. 使用 transactional outbox pattern。

Outbox 会在同一个数据库事务中写入业务结果和待发送事件：

BEGIN;

UPDATE payment_operation
SET status = 'PAYMENT_COMPLETED',
    result_location = :result_location
WHERE operation_id = :operation_id;

INSERT INTO event_outbox (
    event_id,
    aggregate_id,
    event_type,
    payload,
    published
)
VALUES (
    :event_id,
    :operation_id,
    'PaymentCompleted',
    :payload,
    false
);

COMMIT;

另一个 publisher 扫描 outbox，将事件投递到 SQS，然后标记为已发布。即使事件被重复投递，消费者也能通过 eventId 去重。

如果 DocumentDB 的事务边界或现有数据模型不适合实现 outbox，也可以接受”可能重复发布”，但后处理和 invoice 创建必须具备可靠的业务幂等性。

7. 将后处理拆成独立步骤

payment API 完成付款后发布结果事件：

{
  "eventType": "PaymentCompleted",
  "eventVersion": 1,
  "eventId": "evt-01JXYZ",
  "operationId": "pay_01JABC",
  "paymentId": "payment-12345",
  "result": {
    "store": "s3",
    "key": "payments/pay_01JABC/result.json"
  },
  "correlationId": "req-789"
}

Integrator Worker 消费这个事件，执行原有 post processing，再发布：

{
  "eventType": "InvoiceRequested",
  "eventVersion": 1,
  "operationId": "pay_01JABC",
  "paymentId": "payment-12345",
  "invoiceInput": {
    "store": "s3",
    "key": "payments/pay_01JABC/invoice-input.json"
  }
}

这种拆分有几个直接好处：

  • payment API 不需要了解 invoice API；
  • 后处理逻辑仍由 integrator 负责；
  • invoice API 故障不会阻塞 payment worker；
  • 三个阶段可以分别扩容、重试和部署；
  • 将来增加通知、审计等步骤时，不必继续修改 payment API。

如果 invoice API 目前只支持同步 HTTP，可以由 Invoice Worker 消费 SQS 消息后调用它。如果 invoice API 的处理时间也可能很长，则应采用同样的异步模式改造。

8. 重试要区分临时故障和永久故障

并不是所有错误都适合无条件重试。

适合重试的情况包括：

  • 网络超时；
  • 临时连接失败；
  • HTTP 429；
  • HTTP 502、503、504；
  • legacy DBMS 暂时不可用。

通常不适合自动重试的情况包括：

  • payment details 格式错误；
  • 必填字段缺失；
  • 无法识别的账户；
  • 业务规则拒绝；
  • 永久权限错误。

建议为每个队列配置 DLQ，并设置合理的 maxReceiveCount。重试时还应使用指数退避和抖动，避免故障恢复后所有任务同时发起重试：

delay = min(base_delay * (2 ** attempt), max_delay)
delay += random.uniform(0, delay * 0.2)

SQS 对单条消息的延迟能力有限。如果需要数小时级重试、根据错误类型安排不同的重试时间，或者查看每一步的执行历史，可以考虑：

  • 专门的 retry queue；
  • EventBridge Scheduler；
  • Step Functions Standard Workflow；
  • 在 operation 表中保存 nextAttemptAt，再由调度器重新投递。

DLQ 还需要配套告警和重放流程。如果只有 DLQ 而没有监控，故障只是被转移到了另一个队列里。

9. 是否使用 FIFO queue

如果同一个 payment 的消息必须严格有序，可以考虑 FIFO queue，并设置：

MessageGroupId = paymentId

但 FIFO 不能提供端到端的 exactly-once 保证，应用层仍需实现幂等，原因包括：

  • 消费者可能在完成业务操作后、删除消息前崩溃；
  • 去重窗口不是永久的业务约束；
  • 对外部 DBMS 或 invoice API 的调用不在 SQS 事务范围内。

如果不同 payment 之间互不依赖，Standard queue 通常能提供更灵活的吞吐量。不要让所有消息共用同一个 MessageGroupId，否则并发处理会退化为串行处理。

10. 可以使用 Step Functions，但并非必需

如果流程以后会增加许多分支，例如：

付款 → 风控 → 后处理 → 开票 → 通知 → 补偿

Step Functions Standard Workflow 通常比手写状态机更容易观察和维护，尤其适用于：

  • 长时间运行的工作流；
  • 分支和条件判断；
  • 明确的重试、超时和补偿；
  • 人工介入；
  • 查看每一步的执行历史。

大型 payment details 仍不应直接放在工作流状态中传递，只需传递 operationId 或 S3 key。

如果当前流程只有付款、后处理和开票三个稳定步骤，SQS 加 operation 状态表已经足够，不必为了形式上的完整引入额外的编排服务。

11. 建议补齐的运行保障

生产环境至少还应包括：

  • 使用 operationId、eventId、correlationId 串联日志；
  • 监控 queue depth、oldest message age、失败率和处理耗时；
  • 对 DLQ 非空、任务长期停留、临时数据即将过期等情况设置告警；
  • 根据队列积压量扩缩 ECS worker；
  • 为每个步骤设置最大尝试次数和最终失败状态；
  • 使用 IAM role，限制 worker 只能访问所需的 queue、bucket 和 collection；
  • 为 SQS、S3、DocumentDB 以及传输链路启用适当的加密；
  • 不在消息中保存银行卡号、凭证等敏感信息；
  • 为事件增加 eventVersion，便于以后兼容字段变化；
  • 明确定义任务取消、人工重放和部分成功时的处理方式。

推荐方案不只是”在 payment API 前面放一个 queue”，而是把整个请求建模为可追踪的异步 operation。大型数据放在外部存储中，SQS 只传递小型事件，各步骤独立消费，所有副作用都要保持幂等。同时通过 DLQ、过期清理、状态记录和监控处理异常路径。这样才能解决长时间处理带来的问题，避免只是把 HTTP 超时换成更难排查的异步故障。

备注：内容仅供参考。