如何将RabbitMQ队列中存放满5分钟的消息转移到另一队列

下面提供两种不同场景下的实现方式，可根据业务需求选择：

方案1：原生死信交换机（DLX）+ 队列TTL（优先推荐） #

这是RabbitMQ官方提供的原生能力，无需额外开发定时任务，性能最高，适配绝大多数通用场景。
实现步骤：

• 首先声明用来存放超时消息的目标队列，示例命名为timeout_target_queue，按普通队列规则声明即可
• 声明死信交换机，类型选择direct即可，示例命名为dlx_timeout_exchange，参考pika代码：

  channel.exchange_declare(exchange='dlx_timeout_exchange', exchange_type='direct', durable=True)
  

• 将目标队列和死信交换机绑定，绑定路由键可以和原业务队列名保持一致：

  channel.queue_bind(
      exchange='dlx_timeout_exchange',
      queue='timeout_target_queue',
      routing_key='origin_biz_queue'
  )
  

• 声明原业务队列时，添加TTL和死信绑定参数，统一设置队列内所有消息的过期时间为5分钟（300000毫秒）：

  channel.queue_declare(
      queue='origin_biz_queue',
      durable=True,
      arguments={
          'x-message-ttl': 300000,
          'x-dead-letter-exchange': 'dlx_timeout_exchange',
          'x-dead-letter-routing-key': 'origin_biz_queue'
      }
  )
  

注意事项：

1. 已存在的队列无法直接修改arguments参数，需要先删除旧队列再重新声明，或者在发送消息时单独给消息设置expiration参数覆盖队列全局TTL
2. RabbitMQ的消息过期是懒检查机制，只有消息处于队列头部时才会校验是否过期，如果队列头部有大量未过期消息，后续已超时的消息不会立刻触发转移，对时间精度要求极高的场景需要额外做兼容

方案2：自定义定时扫描（适合有特殊过滤规则的场景） #

如果你的业务需要对超时转移的消息做额外判断（比如只有特定标签的消息才需要转移），可以用自定义逻辑实现：

• 发送消息时，在消息的headers属性中添加send_timestamp字段，存储消息发送时的毫秒级时间戳
• 开发定时任务，执行周期可以设为1分钟/30秒，按需调整：
  1. 拉取原业务队列的消息，关闭自动ack配置
  2. 读取消息headers中的send_timestamp，计算当前时间和发送时间的差值，判断是否达到5分钟
  3. 符合转移条件的消息，手动发送到目标队列，然后给原队列的这条消息发送ack确认，删除原队列内的对应消息
  4. 不符合条件的消息发送nack，设置requeue为true，让消息回到原队列继续等待

注意事项：该方案在队列消息量较大时扫描性能会明显下降，没有特殊业务规则要求不推荐使用

内容的提问来源于stack exchange，提问作者user14514318