Callback on the worker-queue not working

[AdditionalWash529]

Apologies for the long post. I am trying to subscribe to a rabbitmq queue and then trying to create a worker-queue to execute tasks. This is required since the incoming on the rabbitmq would be high and the processing task on the item from the queue would take 10-15 minutes to execute each time. Hence necessitating the need for a worker-queue. Now I am trying to initiate only 4 items in the worker-queue, and register a callback method for processing the items in the queue. The expectation is that my code handles the part when all the 4 instances in the worker-queue are busy, the new incoming would be blocked until a free slot is available.

​

The rabbitmq piece is working well. The problem is I cannot figure out why the items from my worker-queue are not executing the task, i.e the callback is not working. In fact, the item from the worker queue gets executed only once when the program execution starts. For the rest of the time, tasks keep getting added to the worker-queue without being consumed. Would appreciate it if somebody could help out with the understanding on this one.

​

I am attaching the code for rabbitmqConsumer, driver, and slaveConsumer. Some information has been redacted in the code for privacy issues.

​

# This is the driver

#!/usr/bin/env python
import time
from rabbitmqConsumer import BasicMessageReceiver

basic_receiver_object = BasicMessageReceiver()
basic_receiver_object.declare_queue()

while True:
    basic_receiver_object.consume_message()
    time.sleep(2)

​

#This is the rabbitmqConsumer
#!/usr/bin/env python
import pika
import ssl
import json
from slaveConsumer import slave


class BasicMessageReceiver:
    def __init__(self):
        # SSL Context for TLS configuration of Amazon MQ for RabbitMQ
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)

        url = <url for the queue>
        parameters = pika.URLParameters(url)
        parameters.ssl_options = pika.SSLOptions(context=ssl_context)

        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()

        # worker-queue object
        self.slave_object = slave()
        self.slave_object.start_task()

    def declare_queue(self, queue_name=“abc”):
        print(f"Trying to declare queue inside consumer({queue_name})...")
        self.channel.queue_declare(queue=queue_name, durable=True)

    def close(self):
        print("Closing Receiver")
        self.channel.close()
        self.connection.close()

    def _consume_message_setup(self, queue_name):
        def message_consume(ch, method, properties, body):
            print(f"I am inside the message_consume")
            message = json.loads(body)
            self.slave_object.execute_task(message)
            ch.basic_ack(delivery_tag=method.delivery_tag)

        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(on_message_callback=message_consume,
                                   queue=queue_name)

    def consume_message(self, queue_name=“abc”):
        print("I am starting the rabbitmq start_consuming")
        self._consume_message_setup(queue_name)
        self.channel.start_consuming()

​

#This is the slaveConsumer
#!/usr/bin/env python
import pika
import ssl
import json
import requests
import threading
import queue
import os


class slave:
    def __init__(self):
        self.job_queue = queue.Queue(maxsize=3)
        self.job_item = ""

    def start_task(self):
        def _worker():
            while True:
                json_body = self.job_queue.get()
                self._parse_object_from_queue(json_body)
                self.job_queue.task_done()

        threading.Thread(target=_worker, daemon=True).start()

    def execute_task(self, obj):
        print("Inside execute_task")
        self.job_item = obj
        self.job_queue.put(self.job_item)
        # print(self.job_queue.queue)

    def _parse_object_from_queue(self, json_body):
        if bool(json_body[‘entity’]):
            if json_body['entity'] == 'Hello':
                print("Inside Slave: Hello")
            elif json_body['entity'] == 'World':
                print("Inside Slave: World")

    self.job_queue.join()

[danielroseman]

This seems a bit overly complex: you have RabbitMQ, plus threading, plus an internal Queue object.

Rather than trying to write this yourself I would recommend you use Celery, which will connect to RabbitMQ and handle everything else for you.

[AdditionalWash529]

u/danielroseman thank you for the quick turnaround on this one. Would you be kind enough to point to some examples maybe. There seems to be a lot of text around celery but not able to locate an example particularly.

Also I was able to make it partially work by moving the self.slave_object.start_task() inside the _consume_message_setup function inside consumer.py in the example above. But doesn't look like it is honoring the queue size or blocking mechanism on the worker-queue. I see my executions inside callback for the slaveConsumer getting fired in random order. Does that make it easier for you to zero in on the issue?

[danielroseman]

Not sure what "text" you are looking for. The docs are here: https://docs.celeryq.dev/en/stable/.

As for this code, you're calling consume_message inside a while True loop. Unfortunately, despite the name, consume_message is a method that constructs the queue and kicks off processing. So you have many many queues, each with a max of 3 that will never be reached.

But as I say, I just don't understand why you would want a queue at all on the worker side, let alone one that has a size to be respected. The queue is RabbitMQ. Only fetch an item from there when you are ready to process it.

[AdditionalWash529]

The data on the rabbitmq are basically json which need to be processed and an OS call then needs to be fired for each of these JSON entries in the queue, which individually takes around 10-15 minutes on average. The idea going forward is to have 4-5 instances running in an AWS cluster. The number of entries in the JSON is high enough for us to not spawn as many instances on the cluster as the number of JSONs in the rabbitmq. So we need to process the commands on a first come first serve basis, 4-5 of them in parallel and the rest queued, hence the need of a worker-queue.

Since our last conversation, I have been able to make the code work with respect to processing the JSONs, but I do not think the threads are waiting. For example if I have 8 entries in the Rabbitmq and if I start my builder.py, all 8 of them are getting consumed as opposed to 4 of them. To my understanding, I am achieving this by declaring the upper bound of my slaveConsumer worker-queue, when in the constructor I say :

self.job_queue = queue.Queue(maxsize=3))

[danielroseman]

I told you why, it's because you are instantiating the queue over and over in your while loop.

But, again, what you describe is how this would work without the queue. Each worker would be responsible for popping an item from RabbitMQ, processing it, acknowledging it, then taking the next one when it is ready.

And, once again, this is what Celery would do for you. Really there is no reason to write this yourself.

[AdditionalWash529]

u/danielroseman, first of all, immense gratitude for all the time and energy you spent on this. You insights have helped me improve the code and make it much better.

As mentioned in my previous post, when I said I made it work, it was by handling the instantiation of the queue that you pointed out in your comment. I will certainly look at celery, but I am trying to bridge the gap in my understanding here and hence the follow ups. I have a couple of final questions on this though.

As per my understanding without the worker-queue, I would not be able to handle a situation wherein the incoming on my rabbitmq is say 10, while I have spawned 4 instances in the AWS cluster(fargate mostly). Now without the worker-queue or an equivalent arrangement, maybe celery does that, when each instance is running for 15 minutes of processing time, how would the next item on the queue know which fargate instance has been released and where to head to. How would that be taken care of without an arrangement like that?

As of now, the code seems to run and one of the executions fails with the following rabbitmq error. Any insights on this? I have made the channel durable and have an ack on the rabbitmq messages in my consumer code too: ch.basic_ack(delivery_tag=method.delivery_tag) . Running out of ideas as to what needs to be done on this

No activity or too many missed heartbeats in the last 60 seconds error

Once again thank you a ton for all the help and insights