Celery queue fetch multiple messages at the same time

user_782400 0
Viewed 645

I am working on a docker-based Celery-backed Python application in which one of the functionality is to trigger and send a text message given a number. The workflow is as follows:

  1. User uploads a CSV with a set of entries to whom the text message should be sent
  2. The cron job polls database every 60 seconds to fetch any new entries and adds them to the queue
  3. If new entries are found, put them in the queue and trigger the text message

Currently, if I upload a CSV file with 3 entries, each text message action is triggered sequentially and not in parallel (default celery process bheaviour). For example, if the scheduler fetches a job from the queue every 10 seconds, the time taken to send 3 text messages will be 30 seconds. Since these jobs are independent of each other, I want to parallelize this such that all three text messages are sent at the same time.

I have tried increasing the concurrency of the queue but assuming that each thread will be assigned one of the three messages but it does not work. I'm afraid there might be something that I am missing. Is there some other configuration I need to add so that I can parallelize the jobs?

Command to run celery queue

celery worker --app=worker.app --concurrency=5 --hostname=worker1@%h --loglevel=INFO-Q queue1 -Ofair

Celery configuration


app = Celery(
    'worker',
    broker=os.environ['CELERY_BROKER'],
    backend=os.environ['RABBITMQ_BACKEND'],
    include=['worker.tasks','worker.schedule']
)



app.conf.update(
    result_expires=3600,
    task_track_started=True,
    worker_prefetch_multiplier = 5
)

app.conf.beat_schedule = {
    "get-message": {
        "task": "worker.schedule.get_new_messages",
        "schedule": 10,
        'options': {'queue' : 'queue1'}
    }
}
docker celery rabbitmq python
2 Answers
user_78977510

The best way to fix this problem is to fix the cron job. In your cron job, instead of putting messages onto the queue, invoke your celery tasks. The reason that messages are being processed sequentially currently depends on your implementation of worker.schedule.get_new_messages. Most likely, that function is pulling more than one message off the queue at a time and the function processing those messages is doing one at a time.

The way to fix that is to make a task that sends one message and that's it. For example:

@app.task('send_my_cool_message')
def send_sms_message(from_, to_, text):
    twilio_client = Client(settings.ACCOUNT_SID, settings.AUTH_TOKEN)
    twilio_client.messages.create(to=to_, from=from_, body=text)

Now in your cron job, you invoke a celery task for each message:

from celery import Celery
qs = Messages.objects.filter(created_at__gte=last_date_polled)
app = Celery(broker=settings.BROKER_URL, backend=settings.BACKEND_URL)
for message in qs:
    app.send_task('send_my_cool_message', kwargs={
        'from_': message.from_,
        'to_': message.to_,
        'text': message.text,
    })
user_60431700
Related