0

作为一个简单的例子,我将 5 个项目添加到一个新的 RabbitMQ(v 2.6.1) 队列中:

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='not.my.real.server.net'))
channel = connection.channel()
channel.queue_declare(queue='dw.neil',durable=True)
# add 5 messages to the queue, the numbers 1-5
for x in range(5):
    message = x+1
    channel.basic_publish(exchange='',routing_key='dw.neil', body=str(message))
    print " [x] Sent '%s'" % message
connection.close()

我清除我的队列,然后运行上面的代码来添加 5 个项目:

nkodner@hadoop4 sports_load_v2$ python send_5.py 
 [x] Sent '1'
 [x] Sent '2'
 [x] Sent '3'
 [x] Sent '4'
 [x] Sent '5'

现在,我正在尝试模拟失败的处理。给定以下代码以从队列中使用。请注意,我对 basic_ack 的调用已注释掉:

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='not.my.real.server.net'))
channel = connection.channel()
channel.queue_declare(queue='dw.neil',durable=True)
method_frame, header_frame, body=channel.basic_get(queue='dw.neil')
print method_frame, header_frame
print "body: %s" % body
#channel.basic_ack(delivery_tag=method_frame.delivery_tag)
connection.close()

我运行接收代码以从队列中获取一个项目。正如我所料,我得到了第 1 项:

nkodner@hadoop4 sports_load_v2$ python r5.py 
<Basic.GetOk(['message_count=9', 'redelivered=False', 'routing_key=dw.neil', 'delivery_tag=1', 'exchange='])>
<BasicProperties([])>
body: 1

由于对 channel.basic_ack() 的调用被注释掉了,我希望将未确认的消息放在队列中,以便下一个消费者得到它。我希望消息 #1 是队列中的第一条消息(再次),并将 Redelivered 属性设置为 True。相反,收到消息 #2:

nkodner@hadoop4 sports_load_v2$ python r5.py 
<Basic.GetOk(['message_count=9', 'redelivered=False', 'routing_key=dw.neil', 'delivery_tag=1', 'exchange='])>
<BasicProperties([])>
body: 2

并且队列中的所有其他消息都在 #1 返回之前接收到,并将 Redelivered 标志设置为 True:

...

nkodner@hadoop4 sports_load_v2$ python r5.py 
<Basic.GetOk(['message_count=9', 'redelivered=False', 'routing_key=dw.neil', 'delivery_tag=1', 'exchange='])>
<BasicProperties([])>
body: 5

nkodner@hadoop4 sports_load_v2$ python r5.py 
<Basic.GetOk(['message_count=9', 'redelivered=True', 'routing_key=dw.neil', 'delivery_tag=1', 'exchange='])>
<BasicProperties([])>
body: 1

是否有任何属性或选项我可以设置,以便我一直收到 #1 直到它被确认?

我的用例是用顺序生成的文件加载数据仓库。我们正在使用基于消息的处理来让我的程序知道一些新文件已准备好并要加载到 DW 中。我们必须按照它们生成的顺序处理这些文件。

4

2 回答 2

3

这已在 RabbitMQ 2.7.0 中得到解决——我们运行的是 2.6.1。

发行说明

此版本中的新功能包括:

  • 为消费者重新排队的消息的保留顺序
于 2012-06-22T20:32:58.647 回答
1

尝试使用 channel.basic_reject - 这应该将未确认的消息推送回 RabbitMQ,RabbitMQ 会将消息视为新消息。另外——如果你有一条失败的消息卡住了,你可以使用 channel.basic_recover 告诉 RabbitMQ 重新传递所有未确认的消息。

http://www.rabbitmq.com/extensions.html#negative-acknowledgements提供了 Basic.Reject 与 Basic.Nack 的区别信息。

消息排序语义在http://www.rabbitmq.com/semantics.html进行了解释

于 2012-05-14T16:01:52.617 回答