CODEWITHSUNDEEP
flaskcelery
X.C.
X.C.

Reputation: 723

Celery task Priority with redis backend in Flask

Below is my simplified website app and celery setup

app = Flask(__name__)
celery = Celery(__name__, broker='redis://localhost:6379/0') 
celery.conf.update({app.config})

and other celery configs are 

CELERYD_MAX_TASKS_PER_CHILD=1 
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT=['json','pickle']

I have two tasks, one is taking a lot of CPU and Memory resources and need a long time to be finished(which I want to set it priority to "low"), and the other one is just to send email for account verification(which I want to set it priority to "high"), defined as below,

@celery.task(serializer='pickle') 
def long_big_task_with_low_priority(...):
    ...   

@celery.task(serializer='pickle') 
def send_email_with_high_priority(...):
    ...

I use supervisord to execute the celery

celery worker -A celerytaskfile.celery --loglevel=error --concurrency=2 --logfile=celerylogfile.log

because the first task consuming lots of resources, therefore I make concurrency 2 here.

My question is when there are two first tasks running, and now celery receives a new task to send email, is it possible to make the send-mail task executed immediately, and then after it completed continue with the first two tasks?

Package Version: Celery:4.1.0 redis==2.10.6

Upvotes: 0

Views: 1477

Answers (1)

zxzak
zxzak

Reputation: 9446

You need two workers. One will only process long_big_task_with_low_priority and the other the rest of the (fast running) tasks. For that you will need a separate queue for the long_big_task_with_low_priority task.

Use CELERY_TASK_ROUTES (CELERY_ROUTES for older celery versions) to route the slow task to a new queue.

CELERY_TASK_ROUTES = {
    'long_big_task_with_low_priority': {'queue': "slow_queue"},
}

When running the workers, assign queues (important for both because otherwise they will consume tasks from any queue)

celery is the default queue name - all tasks go there by default:

celery worker -A celerytaskfile.celery --queues celery

second worker:

celery worker -A celerytaskfile.celery --queues slow_queue

http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-routes https://docs.celeryproject.org/en/latest/userguide/workers.html#queues

Upvotes: 4

Related Questions