Me he decidido a utilizar un sistema de cola de eventos en una aplicación de django. Lo que me ha llevado a tomar esta decisión es fundamentalmente evitar tener que esperar a que un proceso termine para poder seguir interactuando con la aplicación web. Por tanto celery me ofrece justo esto.
Cuando hablamos de celery, siempre viene acompañado del backend RabbitMQ (cola de eventos de alto rendimiento). Pero hay otros sistemas menos complejos, como kombu, el cual utiliza la propia bbdd de django para el almacenamiento de mensajes.
Tenemos la siguiente situación:
- Django 1.3 con una aplicación que lanza procesos sobre diversos servidores
- mod_wsgi para comunicar apache con django
Estos son los dos sistemas que tendremos que tocar para añadir celery a nuestra aplicación.
Tendremos que instalar lo siguiente:
- pip install celery django-kombu django-celery
- apt-get install supervisor ó yum install supervisor (depende del s.o.)
Vamos a decirle a django que vamos a utilizar celery, para ello en nuestro settings.py vamos a añadir lo siguiente:
#Importamos django-celery
import djcelery
djcelery.setup_loader()
#Para evitar problemas a la hora de llamar a las aplicaciones dentro
#INSTALLED_APPS, haremos que tome en el sys.path la ruta actual.
#Por tanto, si la llamada de nuestra aplicación dentro de INSTALLED_APPS la
#hacemos como: misitio.aplicacion, ahora sólo la tendremos que llamar: aplicacion
sys.path.append(os.getcwd())
INSTALLED_APPS = (
....
'djkombu',
'djcelery',
)
#Configuración propia de celery
#Backend a utilizar
BROKER_BACKEND = "djkombu.transport.DatabaseTransport"
#Url de la cola de mensajería, kombu utiliza la bbdd de django
BROKER_URL = "django://"
#Para las tareas programadas, indicamos que utilice la bbdd para su control
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"
Ahora para wsgi, hay que indicarle que vamos a usar celery, por tanto añadimos en nuestro django.wsgi:
import djcelery
djcelery.setup_loader()
Con esto ya hemos acabado de configurar django, ahora actualizaremos nuestar bbdd para que incorpore las tablas necesarias:
python manage.py syncdb
La definición de tareas la haremos dentro de un fichero llamado tasks.py que crearemos dentro de la carpeta de nuestra aplicación. Tenemos la siguiente estructura:
misitio
|-- __init__.py
|-- aplicacion
| |-- __init__.py
| |-- admin.py
| |-- models.py
| |-- tasks.py
| |-- tests.py
| |-- views.py
|-- django.wsgi
|-- manage.py
|-- settings.py
|-- templates
| |-- admin
| | |-- base.html
| | |-- base_site.html
| | `-- index.html
| |
| `-- aplicacion
| |
| `-- index.html
|
`-- urls.py
Vamos a definir una tarea muy simple, y que aparece en todos los ejemplos de celery, editamos el fichero tasks.py y añadimos esto:
from celery import task
@task()
def add(x, y):
return x + y
El decorador @task() será el que indique que la función add será tratada como una tarea asíncrona dentro de celery y que podremos invocar.
Vamos a arrancar celery y a realizar algunas pruebas para comprobar que funciona. Para arrancarlo, de momento vamos a hacerlo del siguiente modo:
python manage.py celeryd -l info --settings=settings
-------------- celery@rafabono v3.0.12 (Chiastic Slide)
---- **** -----
--- * *** * -- [Configuration]
-- * - **** --- . broker: django://localhost//
- ** ---------- . app: default:0xb721c58c (djcelery.loaders.DjangoLoader)
- ** ---------- . concurrency: 4 (processes)
- ** ---------- . events: OFF (enable -E to monitor this worker)
- ** ----------
- *** --- * --- [Queues]
-- ******* ---- . celery: exchange:celery(direct) binding:celery
--- ***** -----
[Tasks]
. aplicacion.tasks.add
Este es el mensaje que nos mostrará al arrancar. Como vemos nos da información del broker (el BROKER_URL que indicamos anteriormente), la concurrencia que ha tomado por defecto (4 procesos), la activación de eventos (que necesitaremos para la tareas programadas) y las colas. Más abajo nos da un listado de las tareas que tenemos en nuestro fichero tasks.py.
Ya que tenemos andando nuestros sistema de cola de eventos, veamos si realmente funciona. En otro teminal abrimos una shell de django (para evitar tener que importar el settings):
python manage.py shell
Y comenzaremos importanto la tarea y haciendo una llamada:
>>> from aplicacion.tasks import add
>>> add.delay(2, 2)
el método delay es el que utilizaremos para llamar a la tarea add con celery. Esto simplemente sumará 2 + 2 pero de forma asíncrona. Tras hacer la llamada veremos lo siguiente:
<AsyncResult: ab56fd91-5ac3-44a8-abee-98d35c6ac537>
Nos devuelve un objeto AsyncResult cuyo identificador es un uuid. La tarea ha sido enviada. Pero, ¿qué ha ocurrido con su ejecución?. Vamos al terminal desde donde lanzamos celery (python manage.py celeryd -l info --settings=settings) y veremos algo así:
[2012-12-05 16:59:18,503: INFO/MainProcess] Got task from broker: aplicacion.tasks.add[ab56fd91-5ac3-44a8-abee-98d35c6ac537]
[2012-12-05 16:59:18,599: INFO/MainProcess] Task aplicacion.tasks.add[ab56fd91-5ac3-44a8-abee-98d35c6ac537] succeeded in 0.0683989524841s: 4
Vemos como la tarea ha llegado a la cola y ha sido ejecutada, dando el resultado esperado: 4
¿Qué más cosas nos puede decir la tarea? Vamos a volver a hacer una llamada, pero esta vez pasándola a una variable:
>>> resultado = add.delay(4, 4)
>>> resultado.ready()
ready() nos dirá si la tarea ha finalizado. La mejor prueba que podemos hacer es para celery (un crtl + c desde el terminal donde lo lanzamos) y volver a hacer una llamada:
>>> resultado = add.delay(4, 4)
>>> resultado.ready()
False
La tarea no ha finalizado, pues la cola está parada. Por tanto esta tarea quedará encolada hasta que la cola vuelva a activarse. La volvemos a activar:
python manage.py celeryd -l info --settings=settings
y veremos como inmediatamente nos aparece el siguiente mensaje:
[2012-12-05 17:12:20,603: INFO/MainProcess] Task aplicacion.tasks.add[ee03a68a-0eb4-43c4-a98a-b3ba72fb5c8c] succeeded in 0.024453163147s: 8
si volvemos a preguntar por su estado:
>>> resultado.ready()
True
También podemos ver el resultado de la tarea:
>>> resultado.result
8
Y ver si ha finalizado correctamente:
>>> resultado.successful()
True
La aplicación de todo esto: infinitas posibilidades. Cada cual que investigue y pruebe para adaptarlo a sus necesidades.
Si en vez de una tarea asíncrona queremos una tarea periódica, la forma de definirla será (por ejemplo):
@periodic_task(run_every=crontab(hour="*", minute="*/5", day_of_week="*"))
def prueba():
print "Hola"
Esta tarea se ejecutará cada 5 minutos, imprimiendo un "Hola".
En cuanto a la forma de lanzar celery, para hacer pruebas siempre podemos hacerlo como hasta ahora, pero en un sistema productivo, habría que hacerlo de forma más segura. Una forma de hacerlo es con el servicio supervisor. Al comienzo de esta entra ya lo instalamos, por lo que tan sólo nos queda configurarlo.
Dependiendo de la version de supervisor, la configuración podrá varia más o menos. A continuación expongo un pequeño ejemplo del fichero supervisord.conf (suele estar en /etc/supervisor o directamente en /etc), sólo de la zona donde definimos celery:
[program:celery]
command=/path/python/bin/python /path/django/misitio/manage.py celeryd -v 2 -B -E -l INFO --settings=settings
process_name = celery
directory=/path/python/bin
user=celery
autostart=true
autorestart=true
priority = 0
stdout_logfile=/var/log/celeryd.log
redirect_stderr=true
Simplemente agregamos una sección para celery, donde indicaremos el comando a lanzar. Tendremos que añadir la opción -B (modo celerybeat para que escuche las peticiones de las tareas programadas) y -E (para los eventos). En el comando habrá que especificar las rutas de python y de nuestro proyecto django.
También le indicamos el usuario con el que se va a lanzar (no es conveniente hacerlo con root).
Con esto será suficiente, sólo nos queda iniciar supervisor (/etc/init.d/supervisor start). Si tenemos habilitado el acceso web podremos ver el estado de nuestro proceso celery.
