Python实现RabbitMQ延迟队列：TTL与DLX原理及代码详解

Cora

发布时间：2026-10-14| 4人浏览过

原创

本文介绍如何利用Python和RabbitMQ实现延迟队列。通过结合消息生存时间（TTL）和死信交换机（DLX）两个核心特性，将过期消息转发至指定队列，从而实现延迟消费。文章详细解析了Queue TTL与Per-Message TTL的区别，并提供了完整的Python生产者与消费者代码示例，解决异步任务中需要延迟处理的场景。

引言与实现思路

在近期的一项任务中，遇到了需要延迟处理的数据场景。最初采用的方案是将数据存入数据库，并编写脚本每隔五分钟扫描数据表进行处理，但实际效果并不理想。鉴于系统已使用RabbitMQ作为异步任务处理的中间件，尝试利用RabbitMQ实现延迟队列。虽然RabbitMQ本身不提供现成的延迟队列插件，但可以通过其两个关键特性来实现：一是消息生存时间（TTL）超时机制，二是死信交换机（DLX）。下文将具体阐述实现原理及代码。

TTL机制原理

延迟队列的基础原理：Time To Live (TTL)

RabbitMQ允许针对Queue设置x-expires参数，或针对Message设置x-message-ttl参数，以控制消息的生存时间。如果消息超时（当两者同时设置时，以先到期的时间为准），该消息将变为死信（dead letter）。

RabbitMQ中设置消息过期时间主要有两种方式：一是通过队列（Queue）的属性进行设置，此时队列中所有消息具有相同的过期时间（本文延迟队列采用的方案）；二是对单条消息单独设置TTL，使得每条消息的过期时间可以不同。

若同时使用上述两种设置，消息的实际过期时间取两者中TTL较小值。一旦消息在队列中的生存时间超过设定的TTL值，该消息即成为死信（dead letter）。

DLX机制原理

Dead Letter Exchanges (DLX)

RabbitMQ的Queue可以配置x-dead-letter-exchange和x-dead-letter-routing-key（可选）两个参数。当队列内出现死信时，系统会依据这两个参数将死信重新路由转发到指定的队列。

x-dead-letter-exchange：指定当出现死信时，将死信重新发送到的目标交换机（exchange）。
x-dead-letter-routing-key：指定当出现死信时，将死信按照指定的路由键（routing-key）进行发送。

队列中出现死信（dead letter）的情况主要包括以下几种：

消息或队列的TTL过期（这是延迟队列所利用的核心特性）。
队列达到最大长度限制。
消息被消费端拒绝（basic.reject或basic.nack）且requeue参数设置为false。

综合上述两个特性，实现延迟队列的逻辑为：将队列设置为TTL规则，当队列TTL过期后，消息变为死信，随后利用DLX特性将其转发至另一个交换机和队列，从而被重新消费，达到延迟消费的效果。

Python实现方案

延迟队列的设计与实现（Python）

根据上述描述，延迟队列的实现大致分为两个步骤：

第一步是产生死信。死信的生成有两种方式：Per-Message TTL和Queue TTL。由于本需求中所有消息的延迟处理时间相同，因此采用Queue TTL设置队列的TTL。若需对队列中的不同消息设置不同的延迟处理时间，则应设置Per-Message TTL（参考官方文档）。

第二步是设置死信的转发规则，即配置Dead Letter Exchanges（参考官方文档）。

代码实现

完整代码如下：


"""
Created on Fri Aug 3 17:00:44 2018

@author: Bge
"""
import pika,json,logging
class RabbitMQClient:
  def __init__(self, conn_str='amqp://user:pwd@host:port/%2F'):
    self.exchange_type = "direct"
    self.connection_string = conn_str
    self.connection = pika.BlockingConnection(pika.URLParameters(self.connection_string))
    self.channel = self.connection.channel()
    self._declare_retry_queue() #RetryQueue and RetryExchange
    logging.debug("connection established")
  def close_connection(self):
    self.connection.close()
    logging.debug("connection closed")
  def declare_exchange(self, exchange):
    self.channel.exchange_declare(exchange=exchange,
                   exchange_type=self.exchange_type,
                   durable=True)
  def declare_queue(self, queue):
    self.channel.queue_declare(queue=queue,
                  durable=True,)
  def declare_delay_queue(self, queue,DLX='RetryExchange',TTL=60000):
    """
    创建延迟队列
    :param TTL: ttl的单位是us，ttl=60000 表示 60s
    :param queue:
    :param DLX:死信转发的exchange
    :return:
    """
    arguments={}
    if DLX:
      #设置死信转发的exchange
      arguments[ 'x-dead-letter-exchange']=DLX
    if TTL:
      arguments['x-message-ttl']=TTL
    print(arguments)
    self.channel.queue_declare(queue=queue,
                  durable=True,
                  arguments=arguments)
  def _declare_retry_queue(self):
    """
    创建异常交换器和队列，用于存放没有正常处理的消息。
    :return:
    """
    self.channel.exchange_declare(exchange='RetryExchange',
                   exchange_type='fanout',
                   durable=True)
    self.channel.queue_declare(queue='RetryQueue',
                  durable=True)
    self.channel.queue_bind('RetryQueue', 'RetryExchange','RetryQueue')
  def publish_message(self,routing_key, msg,exchange='',delay=0,TTL=None):
    """
    发送消息到指定的交换器
    :param exchange: RabbitMQ交换器
    :param msg: 消息实体，是一个序列化的JSON字符串
    :return:
    """
    if delay==0:
      self.declare_queue(routing_key)
    else:
      self.declare_delay_queue(routing_key,TTL=TTL)
    if exchange!='':
      self.declare_exchange(exchange)
    self.channel.basic_publish(exchange=exchange,
                  routing_key=routing_key,
                  body=msg,
                  properties=pika.BasicProperties(
                    delivery_mode=2,
                    type=exchange
                  ))
    self.close_connection()
    print("message send out to %s" % exchange)
    logging.debug("message send out to %s" % exchange)
  def start_consume(self,callback,queue='#',delay=1):
    """
    启动消费者，开始消费RabbitMQ中的消息
    :return:
    """
    if delay==1:
      queue='RetryQueue'
    else:
      self.declare_queue(queue)
    self.channel.basic_qos(prefetch_count=1)
    try:
      self.channel.basic_consume( # 消费消息
        callback, # 如果收到消息，就调用callback函数来处理消息
        queue=queue, # 你要从那个队列里收消息
      )
      self.channel.start_consuming()
    except KeyboardInterrupt:
      self.stop_consuming()
  def stop_consuming(self):
    self.channel.stop_consuming()
    self.close_connection()
  def message_handle_successfully(channel, method):
    """
    如果消息处理正常完成，必须调用此方法，
    否则RabbitMQ会认为消息处理不成功，重新将消息放回待执行队列中
    :param channel: 回调函数的channel参数
    :param method: 回调函数的method参数
    :return:
    """
    channel.basic_ack(delivery_tag=method.delivery_tag)
  def message_handle_failed(channel, method):
    """
    如果消息处理失败，应该调用此方法，会自动将消息放入异常队列
    :param channel: 回调函数的channel参数
    :param method: 回调函数的method参数
    :return:
    """
    channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)

发布消息的代码如下：


from MQ.RabbitMQ import RabbitMQClient
print("start program")
client = RabbitMQClient()
msg1 = '{"key":"value"}'
client.publish_message('test-delay',msg1,delay=1,TTL=10000)
print("message send out")

消费者代码如下：


from MQ.RabbitMQ import RabbitMQClient
import json
print("start program")
client = RabbitMQClient()
def callback(ch, method, properties, body):
    msg = body.decode()
    print(msg)
    # 如果处理成功，则调用此消息回复ack，表示消息成功处理完成。
    RabbitMQClient.message_handle_successfully(ch, method)
queue_name = "RetryQueue"
client.start_consume(callback,queue_name,delay=0)

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。