在开发应用程序时，我发现这非常有用，因为它为您提供了一种仅重新排队消息的替代方法。这可以轻松减少代码的复杂性，并且是RabbitMQ中许多强大隐藏功能之一。
步骤
首先，我们需要设置两个基本通道，一个用于主队列，另一个用于延迟队列。在我最后的例子中，我包括了一些不必要但可以使代码更可靠的额外标记，例如confirm delivery、delivery_mode和durable。您可以在RabbitMQ 手册中找到有关这些内容的更多信息。
在我们设置通道之后，我们向主通道添加一个绑定，以便我们可以使用它将消息从延迟通道发送到我们的主队列。
channel.queue_bind(exchange='amq.direct',
queue='hello')
下一步，我们需要配置延迟通道，使得它在消息过期后将消息转发到主队列。
delay_channel.queue_declare(queue='hello_delay', durable=True, arguments={
'x-message-ttl' : 5000,
'x-dead-letter-exchange' : 'amq.direct',
'x-dead-letter-routing-key' : 'hello'
})
x-message-ttl (消息 - 存活时间)
通常用于在特定持续时间后自动删除队列中的旧消息，但是通过添加两个可选参数，我们可以更改此行为，并且将此参数确定为以毫秒为单位的消息在延迟队列中停留时间。
该变量允许我们在消息过期后将其转移到不同的队列而不是完全删除。
此变量确定用于从hello_delay传输消息到hello队列的交换机。
发布到延迟队列
当我们完成设置所有基本 Pika 参数时，只需使用基本发布将消息发送到延迟队列即可。
delay_channel.basic_publish(exchange='',
routing_key='hello_delay',
body="test",
properties=pika.BasicProperties(delivery_mode=2))
示例。import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))
# Create normal 'Hello World' type channel.
channel = connection.channel()
channel.confirm_delivery()
channel.queue_declare(queue='hello', durable=True)
# We need to bind this channel to an exchange, that will be used to transfer
# messages from our delay queue.
channel.queue_bind(exchange='amq.direct',
queue='hello')
# Create our delay channel.
delay_channel = connection.channel()
delay_channel.confirm_delivery()
# This is where we declare the delay, and routing for our delay channel.
delay_channel.queue_declare(queue='hello_delay', durable=True, arguments={
'x-message-ttl' : 5000, # Delay until the message is transferred in milliseconds.
'x-dead-letter-exchange' : 'amq.direct', # Exchange used to transfer the message from A to B.
'x-dead-letter-routing-key' : 'hello' # Name of the queue we want the message transferred to.
})
delay_channel.basic_publish(exchange='',
routing_key='hello_delay',
body="test",
properties=pika.BasicProperties(delivery_mode=2))
print " [x] Sent"
您可以使用RabbitMQ官方插件：x-delayed-message。
首先，下载并复制ez文件到您的RabbitMQ根目录/plugins中。
其次，启用插件（无需重新启动服务器）：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
最后，使用“x-delay”标头发布您的消息，例如：
headers.put("x-delay", 5000);
注意:
这种方案不能保证您的消息安全，因为如果您的消息在rabbitmq-server宕机期间过期，那么消息将会丢失。因此，在使用此方案时请小心。
祝使用愉快，更多信息请参见rabbitmq-delayed-message-exchange
请注意，如何在Spring 3.2.x中完成此操作。
<rabbit:queue name="delayQueue" durable="true" queue-arguments="delayQueueArguments"/>
<rabbit:queue-arguments id="delayQueueArguments">
<entry key="x-message-ttl">
<value type="java.lang.Long">10000</value>
</entry>
<entry key="x-dead-letter-exchange" value="finalDestinationTopic"/>
<entry key="x-dead-letter-routing-key" value="finalDestinationQueue"/>
</rabbit:queue-arguments>
<rabbit:fanout-exchange name="finalDestinationTopic">
<rabbit:bindings>
<rabbit:binding queue="finalDestinationQueue"/>
</rabbit:bindings>
</rabbit:fanout-exchange>
NodeJS实现。
从代码中可以看出一切都很清楚。 希望能为某些人节省时间。
var ch = channel;
ch.assertExchange("my_intermediate_exchange", 'fanout', {durable: false});
ch.assertExchange("my_final_delayed_exchange", 'fanout', {durable: false});
// setup intermediate queue which will never be listened.
// all messages are TTLed so when they are "dead", they come to another exchange
ch.assertQueue("my_intermediate_queue", {
deadLetterExchange: "my_final_delayed_exchange",
messageTtl: 5000, // 5sec
}, function (err, q) {
ch.bindQueue(q.queue, "my_intermediate_exchange", '');
});
ch.assertQueue("my_final_delayed_queue", {}, function (err, q) {
ch.bindQueue(q.queue, "my_final_delayed_exchange", '');
ch.consume(q.queue, function (msg) {
console.log("delayed - [x] %s", msg.content.toString());
}, {noAck: true});
});
兔子队列中的消息可以通过两种方式延迟： - 使用队列TTL - 使用消息TTL 如果要将队列中的所有消息都延迟固定时间，请使用队列TTL。 如果每个消息都必须延迟不同的时间，请使用消息TTL。 我使用python3和pika模块进行了解释。 pika BasicProperties参数'expiration'（以毫秒为单位）必须设置为延迟消息在延迟队列中。 设置过期时间后，将消息发布到延迟队列（“不是消费者等待消费的实际队列”），一旦延迟队列中的消息过期，消息将通过交换机'amq.direct'路由到实际队列。
def delay_publish(self, messages, queue, headers=None, expiration=0):
"""
Connect to RabbitMQ and publish messages to the queue
Args:
queue (string): queue name
messages (list or single item): messages to publish to rabbit queue
expiration(int): TTL in milliseconds for message
"""
delay_queue = "".join([queue, "_delay"])
logging.info('Publishing To Queue: {queue}'.format(queue=delay_queue))
logging.info('Connecting to RabbitMQ: {host}'.format(
host=self.rabbit_host))
credentials = pika.PlainCredentials(
RABBIT_MQ_USER, RABBIT_MQ_PASS)
parameters = pika.ConnectionParameters(
rabbit_host, RABBIT_MQ_PORT,
RABBIT_MQ_VHOST, credentials, heartbeat_interval=0)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue=queue, durable=True)
channel.queue_bind(exchange='amq.direct',
queue=queue)
delay_channel = connection.channel()
delay_channel.queue_declare(queue=delay_queue, durable=True,
arguments={
'x-dead-letter-exchange': 'amq.direct',
'x-dead-letter-routing-key': queue
})
properties = pika.BasicProperties(
delivery_mode=2, headers=headers, expiration=str(expiration))
if type(messages) not in (list, tuple):
messages = [messages]
try:
for message in messages:
try:
json_data = json.dumps(message)
except Exception as err:
logging.error(
'Error Jsonify Payload: {err}, {payload}'.format(
err=err, payload=repr(message)), exc_info=True
)
if (type(message) is dict) and ('data' in message):
message['data'] = {}
message['error'] = 'Payload Invalid For JSON'
json_data = json.dumps(message)
else:
raise
try:
delay_channel.basic_publish(
exchange='', routing_key=delay_queue,
body=json_data, properties=properties)
except Exception as err:
logging.error(
'Error Publishing Data: {err}, {payload}'.format(
err=err, payload=json_data), exc_info=True
)
raise
except Exception:
raise
finally:
logging.info(
'Done Publishing. Closing Connection to {queue}'.format(
queue=delay_queue
)
)
connection.close()
TTL移动到消息属性中即可。请随意打开一个新问题并在此处链接它，我会回答它。 - eandersson