python - How to cancel conflicting / old tasks in Celery? -

python - How to cancel conflicting / old tasks in Celery? -

i'm using celery + rabbitmq. when celery worker isn't available tasks waiting in rabbitmq. becomes online bunch of tasks executed immediately. can somehow prevent happening?

for illustration there 100 tasks (the same) waiting celery worker, can execute 1 of them when celery worker comes online?

since tasks same in queue, improve way send task once, need able track task published, example:

using lock, example: ensuring task executed 1 @ time

using custom task id , custom state after task published, example:

to add together custom state when task published:

from celery import current_app celery.signals import after_task_publish @after_task_publish.connect def add_sent_state(sender=none, body=none, **kwargs): """track published tasks.""" # task instance name task = current_app.tasks.get(sender) # if there no task.backend fallback app.backend backend = task.backend if task else current_app.backend # store task state backend.store_result(body['id'], none, 'sent')

when want send task can check if task has been published, , since we're using custom state task's state won't pending when it's published (which unkown) can check using:

from celery import states # task has custom id task = task_func.asyncresult('custom_id') if task.state != states.pending: # task exists else: # send task task_func.apply_async(args, kwargs, task_id='custom_id')

i'm using approach in app , it's working great, tasks sent multiple times , identified ids way each task sent once.

if you're still want cancel tasks in queue can use:

# import celery instance project.celery import app app.control.purge()

check celery faq how purge waiting tasks ?

python locking rabbitmq celery