¿Cómo saber si una tarea ya se ha puesto en cola en django-apio?

Question

Aquí está mi configuración:

• Django 1.3
• apio 2.2.6
• django-apio 2.2.4
• djkombu 0.9.2

En mi archivo me settings.py tiene

BROKER_BACKEND = "djkombu.transport.DatabaseTransport" 

es decir, solo estoy usi ng la base de datos para cola de tareas.

Ahora a mi problema: Tengo una tarea iniciada por el usuario que podría tomar unos minutos en completarse. Quiero que la tarea solo se ejecute una vez por usuario, y almacenaré en caché los resultados de la tarea en un archivo temporal, por lo que si el usuario inicia nuevamente la tarea, simplemente devuelvo el archivo en caché. Tengo un código que se ve así en mi función de vista:

task_id = "long-task-%d" % user_id 
result = tasks.some_long_task.AsyncResult(task_id) 

if result.state == celery.states.PENDING: 
    # The next line makes a duplicate task if the user rapidly refreshes the page 
    tasks.some_long_task.apply_async(task_id=task_id) 
    return HttpResponse("Task started...") 
elif result.state == celery.states.STARTED: 
    return HttpResponse("Task is still running, please wait...") 
elif result.state == celery.states.SUCCESS: 
    if cached_file_still_exists(): 
     return get_cached_file() 
    else: 
     result.forget() 
     tasks.some_long_task.apply_async(task_id=task_id) 
     return HttpResponse("Task started...") 

Este código casi funciona. Pero me encuentro con un problema cuando el usuario recarga rápidamente la página. Hay un retraso de 1 a 3 segundos entre el momento en que se pone en cola la tarea y cuando la tarea finalmente se retira de la cola y se entrega a un trabajador. Durante este tiempo, el estado de la tarea permanece PENDIENTE, lo que hace que la lógica de vista inicie una tarea duplicada.

Lo que necesito es una forma de saber si la tarea ya se ha enviado a la cola para que no termine enviándola dos veces. ¿Hay una forma estándar de hacer esto en el apio?

Answer 1

Puede hacer trampas almacenando el resultado manualmente en la base de datos. Déjame explicarte cómo esto ayudará.

Por ejemplo, si se utiliza RDBMS (tabla con columnas - task_id, estatales, resultado):

Ver parte: gestión de transacciones

1. uso.
2. Use SELECCIONAR PARA ACTUALIZAR para obtener la fila donde task_id == "long-task-% d"% user_id. SELECT FOR UPDATE bloqueará otras solicitudes hasta que COMMIT o ROLLBACKs lo hagan.
3. Si no existe, configure estado en PENDIENTE y comience la 'cierta_tarea_longitud', finalice la solicitud.
4. Si el estado es PENDIENTE, informe al usuario.
5. Si el estado es EXITO: ​​establezca el estado en PENDIENTE, comience la tarea, devuelva el archivo al que apunta la columna 'resultado'. Baso esto en la suposición de que desea volver a ejecutar la tarea para obtener el resultado. COMPRAR
6. Si el estado es ERROR - establecer estado en PENDIENTE, comience la tarea, informe al usuario. COMMIT parte

Tarea:

1. preparar el archivo, envolver en intento, bloque catch.
2. En caso de éxito: ACTUALIZAR la fila correcta con estado = SUCCESS, resultado.
3. En caso de falla - ACTUALIZAR la fila correcta con estado = ERROR.

Answer 2

Resolvé esto con Redis. Simplemente configure una clave en redis para cada tarea y luego elimine la clave de redis en el método after_return de la tarea. Redis es liviano y rápido.

Answer 3

No creo (como lo han sugerido Tomek y otros) que usar la base de datos es la forma de hacer este bloqueo. django tiene un marco de caché incorporado, que debería ser suficiente para lograr este bloqueo, y debe ser más rápido. Ver:

http://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html#cookbook-task-serial

Django puede ser configurado para utilizar memcached como motor de caché, y esto puede ser distribuido a través de múltiples máquinas ... esto me parece mejor. ¿Pensamientos?