How to publish messeages to RabbitMQ by using Multi threading?

[atayziv]

I would appreciate help on parallelism with RabbitMQ. Given a large json file, let's say it has 1 million items, I want to go through it in parallel so that each item is passed to Rabbit using multithreading. I read quite a few blogs and it seems that there is a problem with using one connection with one channel with multithreading and if I understood correctly it was recommended to add channels, in practice I didn't really understand how this was implemented in the code, I tried all kinds of things and nothing worked. I would like to know if my desire is possible, or if I have to parse the file iteratively one after the other, and if it is possible if I can get help with the code?

Here i define the Connection and the publish to rabbit function :

class RabbitMQClient:
    def __init__(
        self,
        queues_connection_config: Dict[str, Any],
        queue_name: str,
    ) -> None:
        self.__logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}")
        self.__queue_name = queue_name
        self.__exchange = queues_connection_config.get("exchange")

        self._connection = Connection(
            queues_connection_config.get("hostname"),
            queues_connection_config.get("username"),
            queues_connection_config.get("password"),
            queues_connection_config.get("port"),
            virtual_host=queues_connection_config.get("virtual_host"),
        )
        self.channel=self._connection.channel()
        self.channel.confirm_deliveries()
    
    def enqueue(self, queues_item: Dict[str, Any]) -> str:
        message = Message.create(self.channel, json.dumps(queues_item))
        message.publish(routing_key=self.__queue_name, exchange=self.__exchange, mandatory=True)
        self.__logger.debug(f"Enqueued {message.correlation_id}")
        return str(message.correlation_id)

Here i try to divide the task into multi threading without success:

for i in range(checkpoint, 10 + 1):
                    batch_data = [next(items) for _ in range(1000)]
                    try:
                        start_time = time.time()
                        futures = [
                            self.executor.submit(self.__queue_client.enqueue, item)
                            for item in batch_data
                        ]
                        # Wait for all threads to complete
                        concurrent.futures.wait(futures)

[lukebakken]

Hello, thanks for using RabbitMQ.

If you would like assistance, at the bare minimum, you must provide a git repository with code I can clone and run. Right now you are asking for free assistance using a couple code snippets. We do not have the time to assist with that.

If you've "tried all kinds of things and nothing worked" then at least share something you have tried.

[atayziv]

Dear Luke,

Thank you for your response. I've recently transitioned from using AMQPstorm to Pika for RabbitMQ communication. As part of this transition, I've implemented asynchronous message publishing similar to the example provided in the Pika documentation:

Asynchronous Publisher Example

However, I'm now facing a new challenge. I have a large JSON file containing approximately 100,000 items. For each item in this file, I need to construct an object based on a predefined model.

Could you provide guidance on how I can adapt the asynchronous publishing example to iterate over the large JSON file and process each item accordingly?

Your assistance in this matter would be greatly appreciated.

[lukebakken]

Could you provide guidance on how I can adapt the asynchronous publishing example to iterate over the large JSON file and process each item accordingly

As before, I won't assist you unless you share the code you have written. Team RabbitMQ's community support is not a code-writing service.

[atayziv]

Okay, I am attaching a link to my repository where the relevant rabbit code section is attached.
From another model, the data from Jason is transferred to this function:
initialize_async_publishing
https://github.com/atayziv/RabbitMQ-Paralleled-Publisher/blob/main/async_publish_rabbitMQ.py

[lukebakken]

Please continue discussion here - atayziv/RabbitMQ-Paralleled-Publisher#1

[michaelklishin]

@atayziv on what grounds are you eligible for community support from the core team? Are you a paying customer? Then use a commercial support channel. Are you a regular contributor? Then we would know that.

Our team does not have the resources to troubleshoot everyone's apps. You get the software for free, you get the updates for free, your get client libraries for free, you get the docs for free and — usually — at least some basic guidance from the core team to know where to look next. But it is on you to prove that there is a problem in the client.

[michaelklishin]

Concurrent publishing on a shared channel is explicitly prohibited by all RabbitMQ clients our team maintains. Which means you must use separate channels or synchronize the publishing and publisher confirm operations.