一次异步批量更新500个商品，接口返回bulk_uuid，后台轮询最终显示Complete，但只有460个商品真的更新。这里最容易误解的是“Complete”：它可能表示这批operation已经处理完，并不等价于每条都成功。正确监控必须下钻到operation级别。

保存bulk_uuid以后查每条状态

不要只记录提交接口的HTTP 202。调用Bulk状态接口，或者直接在只读查询中关联magento_bulk与magento_operation。表字段因版本会有差异，先DESCRIBE确认。

SELECT b.uuid,b.description,o.id,o.topic_name,o.status,o.result_message
FROM magento_bulk b
JOIN magento_operation o ON o.bulk_uuid=b.uuid
WHERE b.uuid='BULK-UUID'
ORDER BY o.id;

状态码不要凭数字猜含义，使用当前版本OperationInterface常量或接口响应映射。重点收集失败operation的serialized_data标识、result_message和重试次数，注意不要把完整商品敏感数据写到监控平台。

先按失败类型分组

  • 输入校验：必填属性缺失、option不存在、SKU格式不对
  • 业务冲突：URL key重复、库存Source不存在、商品被其他任务更新
  • 基础设施：数据库死锁、搜索服务不可用、队列连接中断
  • 代码异常：插件抛TypeError、序列化类版本不一致

只有基础设施类失败适合自动重试。输入错误如果原样重投，只会再次失败并堵队列。

重试要生成新的幂等任务

给每条业务更新带外部幂等键，例如source_system + batch_id + sku + payload_version。重试时先读取当前商品状态：目标值已经存在就标记成功，不重复保存；仍不一致才重新发布。不要把原operation表的status直接改回待处理，这会破坏审计和重试次数。

idempotency_key = erp:20260930-01:SKU-1001:v3
if current_state_hash == expected_hash: mark reconciled
else: publish a new operation referencing original operation id

Consumer健康也要一起看

bin/magento queue:consumers:list
bin/magento queue:consumers:start async.operations.all --max-messages=500 -vvv

生产环境通常由Supervisor运行，命令行只用于受控诊断。检查queue_message_status是否长期存在IN_PROGRESS，以及同一topic是否有多个不兼容Consumer版本。

一批任务的完成标准

  • bulk已结束
  • operation成功数 + 明确失败数 = 提交数
  • 所有可重试错误完成有限次数重试
  • 不可重试错误已输出可操作的修正清单
  • 抽样读取商品、库存或价格确认最终状态，而不是只看队列表
异步系统的成功必须以业务结果为准。bulk完成是流程信号，operation结果和最终商品状态才是交付信号。