He estado usando Celery por un tiempo, en producción uso RabbitMQ como intermediario y Redis para el back-end en un clúster K8 sin problemas hasta ahora. Localmente, ejecuto una composición docker con algunos servicios (Flask API, 2 trabajadores diferentes, Beat, Redis, Flower, Hasura), usando Redis como intermediario y backend.
No he tenido problemas con esta configuración durante los últimos meses, pero ayer comencé a tener un comportamiento errático al acceder a los resultados de las tareas.
Las tareas se envían a la cola, el trabajador las reconoce y realiza la tarea, pero al consultar el estado de la tarea, a veces obtengo DisabledBackend . Normalmente en la primera solicitud, y luego funciona. No se pudo encontrar un patrón de cuándo funciona y cuándo no, es errático.
Leí en alguna parte que Celery no funcionó muy bien con el servidor integrado de Flass, así que cambié a uWSGI con prácticamente la misma configuración que tengo en producción:
[uwsgi] wsgi-file = app/uwsgi.py callable = application http = :8080 processes = 4 threads = 2 master = true chmod-socket = 660 vacuum = true die-on-term = true buffer-size = 32768 enable-threads = true req-logger = python:uwsgiHe visto una pregunta similar en Django en la que el problema parecía estar en WSGI Mod con Apache, que no es mi caso, pero el comportamiento parece similar. Todas las demás preguntas que he visto estaban relacionadas con una configuración incorrecta del backend, que no es mi caso.
¿Alguna idea sobre lo que podría estar causando esto? Gracias.
Entonces, parece que necesito acceder a AsyncResult solo a través de mi instancia de la aplicación Celery, en lugar de a través de Celery, o pasar la instancia de la aplicación Celery como argumento.
Entonces, esto no funciona:
from celery.result import AsyncResult @app.route('/status/<task_id>') def get_status(task_id): task = AsyncResult(task_id) return task.stateEsto funciona:
from app import my_celery # Your own Celery Application Instance @app.route('/status/<task_id>') def get_status(task_id): task = my_celery.AsyncResult(task_id) return task.stateEsto también funciona:
from app import my_celery from celery.result import AsyncResult @app.route('/status/<task_id>') def get_status(task_id): task = AsyncResult(task_id, app=my_celery) return task.state Supongo que lo que sucede es que al llamar a AsyncResult directamente desde Celery, no accede a las configuraciones de Celery, por lo tanto, cree que no hay un backend configurado para consultar los resultados.
Pero eso solo explicaría la falla completa de la función, y no el comportamiento errático. Supongo que esto se debe a diferentes subprocesos y situaciones en las que la instancia de la aplicación es importante, por lo que Celery la encuentra, aunque no estoy muy seguro.
Realicé un par de pruebas y parece que vuelve a funcionar bien después de cambiar el AsyncResult importado, pero seguiré investigando.