Django Celery task queue

Question

У меня есть приложение, сделанное с django, используя redis и сельдерей для некоторых асинхронных задач. Я использую задачи сельдерея для выполнения некоторых хранимых процедур. Этот SP занимает от 5 минут до 30 минут для полного выполнения (в зависимости от количества записей). Все отлично работает. Но мне нужно иметь возможность выполнять задачи несколько раз. но прямо сейчас, когда я запускаю задачу, а другой пользователь запускает задачу, две задачи выполняются одновременно. Мне нужно, чтобы задача была введена в очередь и выполнялась только при завершении первой задачи. Мой settings.py:

BROKER_URL = 'redis://localhost:6379/0' 
CELERY_IMPORTS = ("pc.tasks",) 
CELERY_ACCEPT_CONTENT = ['json'] 
CELERY_TASK_SERIALIZER = 'json' 
CELERY_RESULT_SERIALIZER = 'json' 
CELERY_RESULT_BACKEND='djcelery.backends.cache:CacheBackend' 

tasks.py

from __future__ import absolute_import 
from celery.decorators import task 
from celery import task, shared_task 
from .models import Servicio, Proveedor, Lectura_FTP, Actualizar_Descarga 
from .models import Lista_Archivos, Lista_Final, Buscar_Conci 

@task 
def carga_ftp(): 
    tabla = Proc_Carga() 
    sp = tabla.carga() 
    return None 

@task 
def conci(idprov,pfecha): 
    conci = Buscar_Conci() 
    spconc = conci.buscarcon(idprov,pfecha) 

Я называю задачи по моему мнению, таким образом:

conci.delay(prov,DateV); 

Как я могу создать или настроить список очереди таков и вечных задач выполняется только при завершении предыдущих taks.

Заранее благодарен

Answer 1

Вы можете ограничить работу рабочих, в вашей причине я предполагаю, что вам нужен только один работник вовремя, поэтому просто запускайте одного работника при вызове djcelery.

python manage.py celery worker -B --concurrency=1 

Answer 2

Вы можете использовать блокировку, например (от одного из моих проектов):

def send_queued_emails(*args, **kwargs): 
    from mailer.models import Message 
    my_lock = redis.Redis().lock("send_mail") 

    try: 
    have_lock = my_lock.acquire(blocking=False) 
    if have_lock: 
     logging.info("send_mail lock ACQUIRED") 
     from celery import group 

     if Message.objects.non_deferred().all().count() > 0: 
      t = EmailSenderTask() 
      g = (group(t.s(message=msg) for msg in Message.objects.non_deferred().all()[:200]) | release_redis_lock.s(lock_name="send_mail")) 
      g() 
     else: 
      logging.info("send_mail lock RELEASED") 
      my_lock.release() 
    else: 
     logging.info("send_mail lock NOT ACQUIRED") 

    except redis.ResponseError as e: 
     logging.error("Redis throw exception : {}".format(e)) 
    except: 
    my_lock.release() 
    logging.error("send_mail lock RELEASED because of exception")