Subscribing to multiple queues, is it possible to fetch 1 message at a time?

[jonathan-fileread]

Hi All! I'm currently POCing faststream with rabbitMQ. I'm trying to subscribe to multiple queues within my app worker like this

RabbitBroker(
            rabbitmq_broker,
            default_channel=Channel(prefetch_count=prefetch_count),
            client_properties=RabbitClientProperties(heartbeat=heartbeat),
            graceful_timeout=graceful_timeout
        )

for queue in queues:
        queue = RabbitQueue(
            name=queue, passive=True, durable=True, queue_type=QueueType.QUORUM
        )

        @broker.subscriber(queue, no_ack=False, retry=False)
        async def _handler(
            msg: RabbitMessage,
            queue: str = queue,
        ):
            # graceful exit scope and process message
            await handle_message(
                message=msg,
                queue=queue,
                callable_map=callable_map,
            )

    app = FastStream(service.rabbitmq_broker)

    logger.info("Starting FastStream run loop...")
    await app.run()

for each queue, i'm hooking up a subscriber handler, however each subscriber, even with a prefetch_count of 1, will poll 1 message from each queue to process asynchronously. I.e. if I have 3 queues subscribed in that app, it will poll 3 messages to process at once. I'd like to retain subscribing multiple queues in my app - however I'd like to unack/in-transit/process only 1 message at a time in 1 queue.

Is there a best practice way to do this? Here are some solutions I've thought of, but don't go all the way.

• using Asyncio.semaphores to process 1 message at a time (However it looks to still unack/in-transit 1 message per queue, while processing 1 message at a time)
• polling from 1 queue, with an exchange containing a list of routing keys (This polls from 1 queue, with routing keys / exchange pubbing messages to 1 queue. However since RMQ queues are FIFO, this doesn't work well with our app functionality as we don't have message priority implemention)

[Lancetnik]

@jonathan-fileread you should create a separated channel for each subscriber and set qos for it to 1
#2171

[Lancetnik]

Thanks Lancetnik, correct me if i'm wrong, this would till in-transit 1 message per queue at any given time? I would like to inflight 1 message per app (say there are 1 message per queue, subbed to 12 queues) - so 12 messages total - app should in-transit and be processing 1 message synchronously, while continually checking/polling for all queues.

You can set qos to 1 on broker level: this way you limit consuming by 1 message / per app

Or you can set unique channel for each subscriber and set 1 message / per subscriber (queue) limitation

[jonathan-fileread]

Okay, let me try this out. Thanks so much Lancetnik!

[cabal-daniel]

@Lancetnik could you please elaborate on what you mean by qos? We do not see documentation about it https://faststream.airt.ai/latest/api/faststream/rabbit/RabbitBroker/#faststream.rabbit.RabbitBroker

[jonathan-fileread]

@Lancetnik we set rabbitBroker to this

    self.rabbitmq_broker = RabbitBroker(
        rabbitmq_broker,
        default_channel=Channel(prefetch_count=1),
        client_properties=RabbitClientProperties(heartbeat=120),
        graceful_timeout=600
    )
    
    
    `    for queue in queues:
    queue = RabbitQueue(
        name=queue, passive=True, durable=True, queue_type=QueueType.QUORUM
    )

    @broker.subscriber(queue, no_ack=False, retry=False)
    async def _handler(
        msg: RabbitMessage,
        queue: str = queue,
    ):
        # graceful exit scope and process message
        await handle_message(
            message=msg,
            queue=queue,
            callable_map=callable_map,
        )`

however we still notice that it still in-flights 12 messages at a time (if there are 12 queues subbed on the app) (re your suggestion 1). We'll try suggestion two to define a single fixed channel that is shared between all the queues
this is on latest version of faststream

[Lancetnik]

@Lancetnik could you please elaborate on what you mean by qos? We do not see documentation about it faststream.airt.ai/latest/api/faststream/rabbit/RabbitBroker#faststream.rabbit.RabbitBroker

https://www.rabbitmq.com/docs/consumer-prefetch

[jonathan-fileread]

I tried to use 1 shared channel across all subscribers

    single_channel = Channel(prefetch_count=1)
    broker = service.rabbitmq_broker
    # register subscribers before connecting, each with its own channel
    for queue_name in queues:
        logger.info(f"Using channel {id(single_channel)} for queue {queue_name}")
        queue = RabbitQueue(
            name=queue_name, passive=True, durable=True, queue_type=QueueType.QUORUM
        )
        

        @broker.subscriber(queue, channel=single_channel, no_ack=False, retry=False)
        async def _handler(
            msg: RabbitMessage,
            queue: str = queue_name,
        ):
            async with single_message_processor():
                logger.info(f"Processing message from {queue} on channel {id(single_channel)}")
                # graceful exit scope and process message
                await handle_message(
                    message=msg,
                    queue=queue,
                    callable_map=callable_map,
                )

    app = FastStream(service.rabbitmq_broker)

but this shared channel is still unacking multiple messages (as you can see one of the channel is at 3 unacked message)

[Lancetnik]

@jonathan-fileread sorry, my bad – I missed global option. Fixed in #2297

[jonathan-fileread]

@Lancetnik -

got it - we use quorum queues, I was wondering if there is other workarounds we can use besides global_qos param on channel. Here are the docs

Global QoS
Quorum queues do not support global QoS prefetch where a channel sets a single prefetch limit for all consumers using that channel. If an attempt is made to consume from a quorum queue from a channel with global QoS activated a channel error will be returned.

[Lancetnik]

@jonathan-fileread well, it seems like the only option we have is to use some kind of lock inside the code.

[jonathan-fileread]

got it. thank you for helping @Lancetnik !