当prefetch_count == 1时拒绝并重新排队RabbitMQ任务

Question

假设我有一个包含五个项目的队列：

(tail) E, D, C, B, A (head)

我从这个队列的头部消费消息，但我觉得消息 A 现在不适合处理。我用 reject 把这个项目拒绝掉，并设置 requeue=True，这样队列变成了：

(tail) A, E, D, C, B (head)

接着我消费了 B、C、D 和 E，并对每一个都进行了确认（ack）。现在队列里只剩下 A，我不断地消费它并且一再拒绝，形成了一个无尽的循环。如果有新的非 A 的消息进来，它几乎会立刻被消费掉，然后我又会继续尝试消费 A。

我在 Pika 文档中的 Twisted Consumer 示例上做了一个小修改：

import pika
from pika import exceptions
from pika.adapters import twisted_connection
from twisted.internet import defer, reactor, protocol,task


@defer.inlineCallbacks
def run(connection):

    channel = yield connection.channel()

    exchange = yield channel.exchange_declare(exchange='topic_link',type='topic')

    queue = yield channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)

    yield channel.queue_bind(exchange='topic_link',queue='hello',routing_key='hello.world')

    #yield channel.basic_qos(prefetch_count=1)

    queue_object, consumer_tag = yield channel.basic_consume(queue='hello',no_ack=False)

    l = task.LoopingCall(read, queue_object)

    l.start(0.01)


@defer.inlineCallbacks
def read(queue_object):

    ch,method,properties,body = yield queue_object.get()

    print body

    if body == 'A':
        yield ch.basic_reject(delivery_tag=method.delivery_tag, requeue=True)
    else:
        yield ch.basic_ack(delivery_tag=method.delivery_tag)


parameters = pika.ConnectionParameters()
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
d = cc.connectTCP('hostname', 5672)
d.addCallback(lambda protocol: protocol.ready)
d.addCallback(run)
reactor.run()

问题： 注意以下被注释掉的那一行：

#yield channel.basic_qos(prefetch_count=1)

当我取消注释那一行时，消费者在处理消息 A 时，会在拒绝后立刻再次尝试消费它，完全忽略了队列中可能在后面等待的其他项目。它并没有把被拒绝的项目放到队列的尾部，而是不断地重复尝试，完全阻塞了队列中的其他消息。

如果那一行被注释掉，程序就能正常工作（虽然速度稍慢）。如果那一行存在，并且 prefetch_count > 1，它也能正常工作。设置为正好 1 时，会触发这种行为。

我在拒绝消息 A 时是不是漏掉了什么步骤？还是说 Pika 的预取系统根本就不适合这种边缘情况？

消息处理消息队列 rabbitmq 阻塞消费者模式 pika 预取计数消息拒绝

当prefetch_count == 1时拒绝并重新排队RabbitMQ任务

1 个回答

撰写回答