Rabbitmq Stream ставит в очередь потребителей, которые всегда неактивны ⇐ Python

[Anonymous]

Я использую aio_pika в Python для потребления из очереди потока, и я хочу получить количество потребителей, которые потребляют из очереди потока, но я всегда получаю один и тот же результат: 0 потребителей. Я пробовал: использовать несколько потребителей, использовать pika вместо aio_pika и изменять смещение сообщений потребителя, но результат остался прежним.
После использования пользовательского интерфейса управления RMQ я увидел, что количество потребителей всегда равно ноль, независимо от того, сколько потребителей потребляют, потому что все они считаются неактивными
несмотря на то, что мне удалось потреблять сообщения. Моя цель — каким-то образом сделать потребителей активными, чтобы видеть нужное количество потребителей, потребляющих из очереди потока.
Я добавляю свой код Python и снимок экрана очередь потока в пользовательском интерфейсе управления RMQ.
Мой скрипт:

Код: Выделить всё

import asyncio
import aio_pika
from aio_pika.abc import AbstractIncomingMessage

async def consume_message(message: AbstractIncomingMessage) -> None:
async with message.process():
print(f"Received message: {message.body.decode()}")

async def main():
rabbitmq_url = "amqp://guest:guest@localhost/"

connection = await aio_pika.connect_robust(rabbitmq_url)
async with connection:
channel = await connection.channel()
await channel.set_qos(prefetch_count=5)

queue_name = "stream_queue"
await channel.declare_queue(
queue_name,
durable=True,
arguments={
"x-queue-type": "stream"
}
)

print(f"Listening for messages on queue '{queue_name}'...")

queue = await channel.get_queue(queue_name)
await queue.consume(consume_message)
queue_status = await queue.declare()
**        print(f"Consumer count: {queue_status.consumer_count}")  # Result: "Consumer count: 0"**
await asyncio.Future()

if __name__ == "__main__":
asyncio.run(main())

Очередь потока и неактивный потребитель в пользовательском интерфейсе RMQ:
Очередь потока и неактивный потребитель в пользовательском интерфейсе RMQ
Очередь потока и неактивный потребитель в пользовательском интерфейсе RMQ
Очередь потока и неактивный потребитель в пользовательском интерфейсе RMQ
Очередь потока и неактивный потребитель в пользовательском интерфейсе RMQ
Очередь потока и неактивный потребитель в пользовательском интерфейсе RMQ
p>
Я попробовал объявить очередь потока с сообщениями, использовать ее и публиковать в ней больше сообщений, и, несмотря на то, что потребитель получал сообщения, он все равно считался неактивным, несмотря на то, что получал сообщения.

Подробнее здесь: https://stackoverflow.com/questions/793 ... s-inactive