如何在 RabbitMQ 中重新排队 Sweatshop job？

2026-10-24支付老李

结论

不要在 Sweatshop worker 中使用 sleep，也不要用 RabbitMQ 的普通 requeue 实现延迟重试。RabbitMQ 的 basic.nack / basic.reject 配合 requeue: true 时，消息会立即回到队列，很容易造成高频重试循环。

更稳妥的做法是建立一个重试队列：

退款失败后，把消息放入重试队列。
通过 TTL 让消息在重试队列中保留一段时间，例如 3 秒。
TTL 到期后，由 RabbitMQ 通过 Dead Letter Exchange（DLX）将消息重新投递到退款工作队列。
worker 再次处理这条消息。

如果运行环境允许安装插件，也可以使用 RabbitMQ Delayed Message Plugin。不过，TTL + DLX 不依赖插件，部署通常更方便。

为什么不应在 worker 中 `sleep`

下面的写法不会持续占用 CPU 空转，但会一直占着一个 worker：

sleep 3
RefundWorker.async_process_refund(job)

如果 payment processor 长时间故障，大量退款任务会同时进入休眠状态，随之产生几个问题：

worker 的并发槽位会被占满；
正常任务无法及时处理；
如果进程重启，尚未重新发布的任务可能丢失；
每次失败都会创建新消息，重试次数难以控制；
多次投递可能引发重复退款。

直接执行下面的代码也解决不了延迟重试问题：

delivery_info.channel.basic_nack(
  delivery_info.delivery_tag,
  false,
  true
)

requeue: true 不包含延迟语义。队列中的消息较少时，同一条消息可能马上被原 consumer 或其他 consumer 再次取走。

使用 TTL 和 DLX 建立重试队列

消息可以按下面的路径流转：

refund.work
    ↓ 处理失败
refund.retry.3s
    ↓ TTL 到期
refund.work

下面用 Bunny 演示 RabbitMQ 的配置方式。Sweatshop 能否直接取得 channel、delivery tag 和消息属性，取决于具体版本及其封装方式，因此接入时需要按照实际的 Sweatshop consumer API 调整。

channel = connection.create_channel

work_exchange = channel.direct(
  "refund.work.exchange",
  durable: true
)

retry_exchange = channel.direct(
  "refund.retry.exchange",
  durable: true
)

work_queue = channel.queue(
  "refund.work",
  durable: true
)

retry_queue = channel.queue(
  "refund.retry.3s",
  durable: true,
  arguments: {
    "x-message-ttl" => 3_000,
    "x-dead-letter-exchange" => "refund.work.exchange",
    "x-dead-letter-routing-key" => "refund"
  }
)

work_queue.bind(work_exchange, routing_key: "refund")
retry_queue.bind(retry_exchange, routing_key: "refund.retry")

退款失败后，不要立即把原消息放回工作队列，而应将任务参数发布到重试队列：

def process_refund(payload, delivery_info, properties)
  job = JSON.parse(payload)

  result = PaymentProcessor.refund(job.fetch("payment_id"))

  if result.success?
    mark_refund_completed(job.fetch("refund_id"))
    delivery_info.channel.ack(delivery_info.delivery_tag)
    return
  end

  schedule_retry(job, properties)
  delivery_info.channel.ack(delivery_info.delivery_tag)
end

def schedule_retry(job, properties)
  retry_count = properties.headers.fetch("retry-count", 0).to_i + 1

  raise "Refund retry limit exceeded" if retry_count > 5

  RETRY_EXCHANGE.publish(
    JSON.generate(job),
    routing_key: "refund.retry",
    persistent: true,
    content_type: "application/json",
    headers: {
      "retry-count" => retry_count
    }
  )
end

这里要注意确认顺序：重试消息成功发布后，才能确认原消息。实际项目中还应启用 publisher confirms，降低因重试消息发布失败而原消息已经确认所造成的任务丢失风险。

让工作队列自动死信

也可以直接为退款工作队列配置 DLX。遇到可重试错误时，worker 对原消息执行：

channel.basic_nack(
  delivery_info.delivery_tag,
  false,
  false
)

最后一个参数必须是 false。这样，消息不会立即回到原队列，而会进入该队列配置的 Dead Letter Exchange，再经过重试队列的 TTL 返回工作队列。

这种方式不需要应用重新发布消息，但有几个地方需要处理：

RabbitMQ 会在消息头中维护 x-death 信息，可以据此判断死亡和重试次数；
超过最大重试次数后，应将消息路由到最终失败队列，不能继续循环；
DLX 的具体路由由队列参数或 RabbitMQ policy 决定；
如果 Sweatshop 会自动确认消息，需要先改为手动确认模式，否则无法可靠使用 nack。

实现退避重试

每次固定等待 3 秒通常不够合适。payment processor 发生故障时，大量任务会每隔 3 秒同时重试，给外部服务带来更大压力。

可以建立多个 TTL 队列：

refund.retry.5s
refund.retry.30s
refund.retry.5m
refund.failed

例如使用下面的策略：

def retry_routing_key(retry_count)
  case retry_count
  when 1
    "refund.retry.5s"
  when 2
    "refund.retry.30s"
  when 3..5
    "refund.retry.5m"
  else
    "refund.failed"
  end
end

为每个重试队列设置不同的 x-message-ttl，到期后统一死信回 refund.work。生产环境还可以加入随机抖动，避免大量消息同时返回工作队列。

使用队列级 TTL 时，延迟通常需要按队列划分。不要把延迟时间不同的消息随意放进同一个重试队列，否则队首的长延迟消息可能影响后续消息按时处理。

如果可以安装延迟消息插件

RabbitMQ Delayed Message Plugin 可以通过 x-delay 消息头指定延迟时间，使用方式更直接：

delayed_exchange.publish(
  JSON.generate(job),
  routing_key: "refund",
  persistent: true,
  headers: {
    "x-delay" => 3_000,
    "retry-count" => retry_count
  }
)

这种方案依赖额外插件。是否使用，需要结合 RabbitMQ 集群的运维权限、版本兼容性和可用性要求来决定，不能默认所有托管 RabbitMQ 服务都支持安装该插件。

退款任务还必须保证幂等

RabbitMQ 通常提供 at-least-once delivery。即使确认和重试逻辑没有问题，网络中断、进程崩溃或确认丢失仍可能导致重复投递。因此，退款操作不能建立在消息只会执行一次的假设上。

建议为每次退款生成稳定的幂等键：

idempotency_key = "refund:#{job.fetch("refund_id")}"

PaymentProcessor.refund(
  job.fetch("payment_id"),
  idempotency_key: idempotency_key
)

还需要在数据库中记录退款状态，并通过事务或唯一约束防止同一笔退款被重复执行。只有连接超时、限流或服务暂时不可用等暂时性故障才应重试。参数错误、退款已拒绝等永久性错误应直接进入失败处理流程。

推荐的整体方案是手动确认消息，使用 TTL + DLX 进行延迟重试，限制重试次数并采用指数退避，同时保证退款操作幂等。这样既不会让 worker 因休眠而被占用，也不会因消息立即重新投递而形成 busy loop。