Ответ 1
Кажется, я решил это, переместившись туда, где вызывается basic_qos.
Размещение его сразу после появления channel = connection.channel() изменит поведение на то, что я ожидаю.
Pika + RabbitMQ: установка basic_qos для prefetch = 1 все еще, по-видимому, потребляет все сообщения в очереди
У меня есть рабочий клиент python, который объединяет 10 работников, каждый из которых подключается к очереди RabbitMQ. Немного похоже на это:
#!/usr/bin/python
worker_count=10
def mqworker(queue, configurer):
connection = pika.BlockingConnection(pika.ConnectionParameters(host='mqhost'))
channel = connection.channel()
channel.queue_declare(queue=qname, durable=True)
channel.basic_consume(callback,queue=qname,no_ack=False)
channel.basic_qos(prefetch_count=1)
channel.start_consuming()
def callback(ch, method, properties, body):
doSomeWork();
ch.basic_ack(delivery_tag = method.delivery_tag)
if __name__ == '__main__':
for i in range(worker_count):
worker = multiprocessing.Process(target=mqworker)
worker.start()
Проблема заключается в том, что, несмотря на настройку basic_qos на канале, первый рабочий, начинающий принимать все сообщения, покидает очередь, в то время как остальные сидят там без дела. Я вижу это в интерфейсе rabbitmq, что даже когда я установил worker_count равным 1 и дамп 50 сообщений в очереди, все 50 переходят в "неподтвержденное" ведро, тогда как я ожидаю, что 1 станет непризнанным, а другой 49, чтобы быть готовым.
Почему это не работает?