Recursos Oficiales
Sitio web oficial de Celery: http://www.celeryproject.org/
Documentación oficial en inglés: http://docs.celeryproject.org/en/latest/index.html
Documentación oficial en español: http://docs.jinkan.org/docs/celery/
Introducción a Celery
Framework de Tareas Asíncronas
"""
1) No depende de ningún servidor externo, puede iniciarse mediante sus propios comandos
2) El servicio Celery proporciona soluciones de tareas asíncronas para otros proyectos
Nota: Se ejecutan dos servicios simultáneamente: el servicio del proyecto y el servicio Celery.
El proyecto envía las tareas que requieren procesamiento asíncrono al servicio Celery,
que las ejecuta de forma asíncrona cuando es necesario.
Una persona es un servicio independiente | El hospital también es un servicio independiente
Normally, a person can perform all health-related actions without hospital involvement;
but when sick, the hospital receives and solves the health problem.
The treatment plan for illness is handed to the hospital. When no one is sick,
the hospital runs independently; when someone is sick, the hospital solves the problem.
"""
Arquitectura de Celery
La arquitectura de Celery consta de tres componentes principales:
- Intermediario de mensajes (message broker): Celery no proporciona servicios de mensajería por sí mismo, pero se integra fácilmente con intermediarios de terceros como RabbitMQ o Redis.
- Unidad de ejecución de tareas (worker): Worker es la unidad de ejecución proporcionada por Celery, que se ejecuta de forma concurrente en nodos del sistema distribuido.
- Almacén de resultados de tareas (task result store): Almacena los resultados de las tareas ejecutadas por los workers. Celery soporta diferentes almacanamientoes como AMQP o Redis.
Casos de Uso
- Ejecución asíncrona: Resuelve tareas que consumen mucho tiempo
- Ejecución diferida: Resuelve tareas programadas para ejecutarse después
- Ejecución periódica: Resuelve tareas recurrentes (tareas cron)
Instalación y Configuración
pip install celery
Método de Configuración Básica
Se recomienda utilizar como intermediario de mensajes una base de datos cache: RabbitMQ o Redis
"""1. Configurar el objeto Celery como app"""
import celery
broker = 'redis://127.0.0.1:6379/1' # Usar db1 de redis para almacenar
backend = 'redis://127.0.0.1:6379/2' # Usar db2 de redis para almacenar
app = Celery('nombre_tarea', broker=broker, backend=backend)
# broker: almacena el código de la tarea (una db específica de la base de datos cache)
# backend: almacena los resultados de la ejecución
"""2. El decorador app.task registra funciones como tareas"""
@app.task # Registra la tarea en la app, provee el método delay
def procesar(*parametros):
Proceso_ejecutar
return resultado
"""3. Iniciar el worker para ejecutar unidades, escuchando tareas registradas"""
En Windows se requiere instalar el módulo eventlet:
- Versiones anteriores a 4.x:
celery worker -A principal -l info -P eventlet
- Versiones 4.x y posteriores:
celery -A principal worker -l info -P eventlet
-A: nombre del archivo py que contiene la app
-l: nivel de configuración de logging
-P eventlet: requisito para Windows
## Después de ejecutar, el broker configurado almacenará las funciones registradas
"""4. Para invocar una tarea, usar delay para enviar la tarea al worker,
el estado o resultado se almacenará en la base de datos backend"""
procesar(*parametros) # Llamada normal, ejecución síncrona
procesar.delay(*parametros) # Usar delay envía la tarea al broker (aún no se ejecuta)
# Retorna un objeto AsyncResult, al imprimir muestra su atributo id
"""5. Para ver el resultado, usar el objeto AsyncResult devuelto por delay
o crear una nueva instancia usando el valor del id"""
from principal import app
from celery.result import AsyncResult
identificador_tarea = '51611be7-4914-4bd2-992d-749008e9c1a6'
if __name__ == '__main__':
resultado = AsyncResult(id=identificador_tarea, app=app)
resultado.successful() # Verifica si la ejecución fue exitosa
resultado.get() # Si fue exitosa, obtiene el resultado
resultado.failed() # Verifica si la ejecución falló
resultado.status # Estado de ejecución: PENDING, RETRY, STARTED, SUCCESS, etc.
Configuración como Paquete
La configuración básica crea el objeto Celery en un archivo de módulo .py, por lo que el comando worker es:
celery -A nombre_modulo worker -l nivel_log -P eventlet
Si se configura como estructura de paquete, el comando cambia a:
celery -A nombre_paquete worker -l nivel_log -P eventlet
- Paquete ejemplo: tareas_celery
- __init__.py
- celery.py # Debe llamarse celery para el objeto (usa reflexión)
Celery(broker=broker, backend=backend, include=[
'tareas_celery.tarea_pedidos', # Relativo al directorio del paquete
'tareas_celery.tarea_usuarios', # Registra todas las funciones decoradas con app.task
'tareas_celery.tarea_utilidades', # La reflexión resuelve conflictos de orden
'tareas_externas'
])
- tarea_pedidos.py # Diferentes archivos de tareas definen distintas tareas asíncronas
- tarea_usuarios.py
from .celery import app
@app.task
def funcion_ejemplo():...
- tarea_utilidades.py
- ...
- tareas_externas.py # Funciones fuera del paquete también pueden registrarse
from tareas_celery.celery import app
@app.task
def otra_funcion():...
Nota: Las funciones externas pueden registrarse como tareas asíncronas, lo que significa que para agregar una función como tarea, no es necesario reorganizar la estructura de directorios, solo agregar el decorador app.task y registrar en la app de Celery.
Invocación de Tareas Asíncronas y Obtención de Resultados
Envío inmediato de tareas
resultado = funcion.delay(argumentos_necesarios)
# resultado es un objeto AsyncResult, con atributo id y métodos para obtener resultados
Envío diferido de tareas
El parámetro eta especifica el momento de ejecución
resultado = funcion.apply_async(args=[argumentos], eta=objeto_datetime)
# Para configurar el objeto datetime se puede usar timedelta
# o datetime.strptime(cadena_fecha, formato)
Configuración de Tareas Programadas
Las tareas programadas requieren un proceso que envíe continuamente tareas. Se puede implementar mediante un ciclo en Python, pero se recomienda usar el comando beat de Celery para una configuración más completa.
Configuración en la app:
## Archivo celery
app = Celery(...) # Después de crear la app
# 1. Configurar zona horaria UTC+8
app.conf.timezone = "Asia/Shanghai"
app.conf.enable_utc = False
# 2. Configurar beat_schedule
app.conf.beat_schedule = {
'nombre_tarea': {
'task': 'tareas_celery.tarea_usuarios.enviar_sms',
# 'schedule': timedelta(seconds=3), # Intervalo de ejecución
# 'schedule': crontab(hour=8, day_of_week=1), # Cada lunes a las 8:00
'schedule': crontab(hour=9, minute=43), # Diario a las 9:43
'args': ('18888888', '6666'), # Parámetros requeridos
'kwargs':{} # Parámetros adicionales
'relative':True # Ejecuta basado en la hora más cercana
},
}
Iniciar worker y beat:
celery -A tareas_celery worker -l info -P eventlet # Iniciar worker
celery -A tareas_celery beat -l info # Iniciar beat para tareas programadas
Objeto AsyncResult
Es el valor de retorno de los métodos de envío como delay. Contiene métodos para obtener reusltados. Los atributos del objeto son simples, solo el valor id.
- id: Atributo de datos, clave del resultado en memoria
- get(): Obtiene el resultado de la ejecución, no funciona si falló
- successful(): Retorna true si fue exitoso
- failed(): Retorna true si falló
- status: Retorna el estado actual de la tarea: PENDING, RETRY, STARTED, SUCCSESS, etc.
Uso de Celery con Django
Al configurar funciones que usan módulos y configuraciones de Django como tareas asíncronas de Celery, surge un problema: el entorno de trabajo del worker es independiente de Django. Por lo tanto, al enviar tareas,会出现配置缺失的问题. La solución es configurar el entorno Django en celery.py, ya que Celery es el archivo de inicio del worker.
## celery.py
import os
import celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proyecto_api.configuracion_desarrollo')
app = Celery(...)