Python RabbitMQ Concurrency

Question

In Spring, to set the concurrency for rabbitmq-consumer is so easy. Like:

container.setConcurrentConsumers(consumerSize);
container.setMaxConcurrentConsumers(consumerMaxSize);

Is it possible in python?

My python code looks like:

async def handle_message(loop):
    connection = await connect(SETTINGS.cloudamqp_url, loop = loop)

    channel = await connection.channel()

    queue = await channel.declare_queue(SETTINGS.request_queue, durable=True)

    await queue.consume(on_message, no_ack = True)

score 0 · Accepted Answer · answered Jun 18 '22 at 09:29

I solved my problem with using Thread:

My code looks like:

import threading

from aio_pika import connect, IncomingMessage, Message
import json

class QueueWorker(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.connection = None
        self.channel = None
        self.queue = None

    async def init(self, loop):
        self.connection = await connect(cloudamqp_url, loop=loop)
        self.channel = await self.connection.channel()
        await self.channel.set_qos(prefetch_count=1)
        self.queue = await self.channel.declare_queue(queue, durable=True)
        await self.queue.consume(self.callback, no_ack=False)

    async def callback(self, message: IncomingMessage):
        request = json.loads(message.body.decode("utf-8"))
        try:
            handle(request)
        except Exception as e:
            handleException...
        finally:
            await message.ack()

Consume queue with concurrency:

async def listen_queue(loop):
    for _ in range(consumer_count):
        td = QueueWorker()
        await td.init(loop)

Note: Inspired from Consuming rabbitmq queue from inside python threads

Python RabbitMQ Concurrency

1 Answers1