Dans le monde du backend, il arrive toujours un moment où des applications complexes arrivent aux limites matérielles de leur support. Par exemple, si votre application reçoit plus de tâches que ce que votre CPU peut traiter dans le même temps, c’est que vous avez atteint la limite matérielle de votre machine.
Généralement, pour pouvoir sereinement passer ce cap, nous avons besoin de faire appel à la notion de scalabilité. La scalabilité qualifie un système informatique apte à s’adapter d’un point de vue dimensionnel, tant vers des tailles inférieures que vers des tailles supérieures. Cela concerne aussi bien le flux, le volume, l’espace et le temps. Cela signifie que si votre système est scalable et que la capacité de calcul de votre machine a atteint ses limites, vous pouvez rajouter une machine supplémentaire temporaire pour accélérer le traitement et l’enlever quand elle ne sera plus utile.
En python, il existe un Framework open source vraiment très bien fait qui permet de rendre votre système scalable : Celery. Ce Framework est un gestionnaire de file d’attente, de tâches et de tâches asynchrones basée sur la transmission de messages distribuée. Concrètement, on peut résumer son principe de fonctionnement par le schéma suivant :
Entre I/O Bound et CPU Bound
À quel moment pouvons-nous être sûrs que nous avons besoin de sortir la grosse artillerie afin de distribuer des tâches ?
La réponse est simple. Si vos tâches sont I/O bound vous ne vous orienterez pas vers Celery. Une tâche I/O bound signifie que la vitesse à laquelle un processus progresse est limitée par la vitesse du sous-système d’Entrées/Sorties. Une tâche qui traite les requêtes utilisateurs d’un site internet, par exemple, en accédant à des données en base et en retournant le résultat à l’utilisateur, est susceptible d’être limitée par les Entrées/Sorties. Pour répondre aux limitations des tâches I/O bound, vous utiliserez plutôt des coroutines ainsi que la librairie asyncio.
Par contre, si vos tâches sont CPU bound alors Celery vous apportera un gain non négligeable pour bien gérer la montée en charge. Une tâche CPU bound (ou compute bound) signifie que l’état d’un ordinateur lorsque le temps requis pour effectuer une tâche est déterminée principalement par la vitesse de son processeur. L’utilisation du processeur est alors élevée, peut-être à 100%, pendant plusieurs secondes ou plusieurs minutes.
Pourquoi celery ?
Maintenant que nous avons exprimé le besoin, attardons-nous quelques instants sur la solution proposée par le Framework Celery.
Celery n’est pas un Framework jeune. Sa première release date d’avril 2009. Il est là depuis longtemps et il promet de rester avec nous encore pour longtemps. On remarque aussi que les développeurs ont une très bonne réactivité pour corriger les bugs. Concrètement, Celery est un Framework dont la maturité n’est plus à démontrer.
Pour fonctionner correctement, ce Framework a besoin de travailler avec au moins une base de données dédiée. Cette base lui sert pour écrire et transférer des messages entre les tâches ainsi que pour stocker des résultats. À titre informatif, le broker (messager) usité par la communauté est RabbitMQ. Par contre, pour le stockage et le transit des résultats, le backend le plus employé est redis.
Le choix de la combinaison broker/backend est à votre discrétion. Dans tous les cas, Celery est capable d’interagir avec les plus courants :
- RabbitMQ
- Redis
- Amazon SQS
- Memcache
- Apache Cassandra
- Couchbase
- ArangoDB
- Elasticsearch
- Riak
- AWS DynamoDB
- Zookeeper
- SQLAlchemy
- Django
Bien que le broker et le backend apportent leur lot de latence, cet inconvénient est largement contrebalancé par le gain en scalabilité apporté par la solution. Il faudra toutefois toujours garder un œil sur le trafic réseau pour que celui-ci ne devienne pas votre principal goulot d’étrangement.
Pour ce qui est de l’intégration dans votre code, elle est très simple. Celery vous propose des décorateurs dans lesquels vous pourrez définir le nom des tâches, les timeouts et votre police de retries. Rien de plus simple !
Et pour finir, Celery peut aussi être accompagné d’un superviseur de tâche (flower) qui utilise le broker et le backend pour monitorer les tâches. Au moins, vous n’aurez pas à le coder.
Place à la pratique, les prérequis
Avant de commencer, il vous faut sur votre machine au moins une instance de redis et de rabbitmq. Pas de panique, en 10 minutes, selon votre connexion, vous pouvez les avoir prêt à l’emploi grâce à docker.
Télécharger et exécuter un conteneur rabbitmq :
docker run -p 15672:15672 -p 5672:5672 rabbitmq:3-managementSi cela c’est bien passé, en visitant l’url http://127.0.0.1:15672/ (utilisateur : guest / mot de passe : guest) vous accéderez à l’interface de gestion de rabbitmq.
Ensuite, nous pouvons télécharger et exécuter un conteneur redis :
docker run -p 6379:6379 -e ALLOW_EMPTY_PASSWORD=yes redis:latestDans l’interface de docker, vous devriez maintenant voir les 2 images :
Ainsi que les deux conteneurs actifs :
Maintenant que nous avons nos prérequis en termes de broker et de backend, il ne nous reste plus qu’à préparer notre environnement de travail.
Pour cela, démarrons notre environnement virtuel et installons le framework celery, l’outil de monitoring flower et les connecteurs pour redis et rabbitmq :
pip install celery=5.1.2 celery[librabbitmq] celery[redis] flowerÉcrire des tâches
Les points essentiels à retenir pour écrire des tâches avec le framework Celery sont les suivants :
- Définir une application Celery
app = Celery(name="...", ...)avec les informations du broker et backend ; - Décorer chaque fonction que vous voulez transformer en tâche avec le décorateur approprié
@app.task(...); - Donner un nom à vos tâches, par exemple en reprenant le nom de la fonction décorée :
@app.task(name="..."); - Utiliser des variables d’environnement (
os.environ.get(...)) pour récupérer les informations du broker et du backend ; - Dans chaque tâche, définir une logique
try / exceptpour gérer la reprise sur erreur ; - Utiliser la librairie
loggingpour logger toutes les informations utiles.
Voici un exemple minimal illustrant la définition de tâches avec les bonnes pratiques préalablement listées :
#tasks.py
import logging
from os import environ
from celery import Celery
CELERY_BROKER_URL = environ.get('CELERY_BROKER_URL','pyamqp://guest:guest@127.0.0.1:5672//')
CELERY_RESULT_BACKEND = environ.get('CELERY_RESULT_BACKEND','redis://127.0.0.1:6379/0')
app = Celery(name='projet', backend=CELERY_RESULT_BACKEND)
@app.task(name="task_1")
def task_1(message):
""" celery task - task_1 """
try:
return f"task_1 message : {message}"
except Exception as exc:
logging.exception(exc)
self.retry(exc=exc)
@app.task(name="task_2")
def task_2(message):
""" celery task - task_2 """
try:
return f"task_2 message : {message}"
except Exception as exc:
logging.exception(exc)
self.retry(exc=exc)Enfin, nous pouvons maintenant démarrer nos workers (unités de travail). Pour cela, on doit fournir à celery :
- Le nom du projet définit dans
app = Celery(name="..."); - Le chemin vers l’application
apppar rapport au répertoire courant, dans mon castasks.app; - Le nombre de processus concurrents qui pourront prendre des tâches pour chaque
worker; - Le format du nom qui sera alloué à chaque worker pour s’y retrouver dans l’interface de monitoring.
Pour résumer, voici la ligne de commande à exécuter dans votre environnement de travail avec pour répertoire courant le dossier du projet :
celery -A tasks:app worker --concurrency=2 -n worker@%hÀ noter que pour la concurrence, si vous avez une problématique particulière pour la gérer, vous pouvez spécifier le type de Pool à utiliser.
Monitorer vos workers
Notre worker est maintenant opérationnel dans son coin. On peut le vérifier à l’aide de l’interface de monitoring flower.
Pour cela, rien de plus simple, il vous suffit d’ouvrir un nouvel environnement de travail et de faire appel à celery. Sauf que cette fois, vous allez lui demander de charger l’interface de monitoring flower :
celery flower --broker=amqp://guest:guest@127.0.0.1:5672// --broker-api=http://guest:guest@127.0.0.1:15672/api/ --result-backend=redis://127.0.0.1:6379/0 --port=5555Enfin, pour aller voir cette interface, rendez-vous à l’adresse http://127.0.0.1:5555 :
Vous remarquerez qu’il y a bien au moins un worker en vie, ce qui démontre que celui-ci est prêt à recevoir et traiter des tâches.
Écrire un client
Nous voilà dans la dernière étape, il nous faut maintenant un client pour appeler les tâches.
Rien de bien compliqué :
#client.py
from os import environ
from celery import Celery
from tasks import task_1
CELERY_BROKER_URL = environ.get('CELERY_BROKER_URL','pyamqp://guest:guest@127.0.0.1:5672//')
CELERY_RESULT_BACKEND = environ.get('CELERY_RESULT_BACKEND','redis://127.0.0.1:6379/0')
# Celery instance
app = Celery(name='projet', broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND)
# Send task to rabbitMq, wait the result and retrieve it from redis backend
task = task_1.delay("message produit avec la méthode delay")
result = task.get()
print (result)task_1 message : message produit avec la méthode delayEn important dans votre script la fonction task_1, grâce à son décorateur, nous disposons des méthodes ajoutées par celui-ci. Ce sont celles-ci que l’on utilise (.delay() et .get()) pour appeler et récupérer le résultat de la fonction.
Limiter les imports
L’exemple précédent montre que le client qui fait appel à des tâches doit préalablement les avoir importées. Cette pratique n’est pas envisageable dans le cas où votre client est entièrement découplé du code des tâches.
Mais pas de panique, car nous pouvons appeler une tâche simplement en connaissant son nom et ses arguments. On utilise pour cela la méthode send_task() :
#client.py
from os import environ
from celery import Celery
CELERY_BROKER_URL = environ.get('CELERY_BROKER_URL','pyamqp://guest:guest@127.0.0.1:5672//')
CELERY_RESULT_BACKEND = environ.get('CELERY_RESULT_BACKEND','redis://127.0.0.1:6379/0')
# Celery instance
app = Celery(name='projet', broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND)
# Send task to rabbitMq, wait the result and retrieve it from redis backend
task = app.send_task('task_1', args=("message produit avec send_task",), retry=True)
result = task.get()
print (result)task_1 message : message produit avec send_taskAppeler des tâches en série
Celery va très loin dans le fonctionnel et vous offre plein de possibilités, notamment en ce qui concerne le workflow.
Par exemple, vous pouvez exécuter des chaines de tâches dont le résultat de chaque tâche sera automatiquement injectée dans la suivante :
from os import environ
from celery import Celery
from celery import signature
CELERY_BROKER_URL = environ.get('CELERY_BROKER_URL','pyamqp://guest:guest@127.0.0.1:5672//')
CELERY_RESULT_BACKEND = environ.get('CELERY_RESULT_BACKEND','redis://127.0.0.1:6379/0')
# Celery instance
app = Celery(name='projet', broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND)
signature_task_1 = signature('task_1', args=("message qui sera traité par task_1 puis par task_2",))
# Je ne précise pas d'argument car l'argument sera celui de la tâche précédente
signature_task_2 = signature('task_2')
# La tâche 1 est executée et retourne son résultat à la tâche 2 sous forme d'argument
tasks = (signature_task_1 | signature_task_2)()
result = tasks.get()
print (result)task_2 message : task_1 message : message qui sera traité par task_1 puis par task_2Évidemment, si vous pouvez faire une chaine, vous pouvez faire des appels parallèles :
#client.py
import logging
from os import environ
from celery import Celery
CELERY_BROKER_URL = environ.get('CELERY_BROKER_URL','pyamqp://guest:guest@127.0.0.1:5672//')
CELERY_RESULT_BACKEND = environ.get('CELERY_RESULT_BACKEND','redis://127.0.0.1:6379/0')
CELERY_RESULT_TIMEOUT = 60
# celery instance: 1 is load in each process pool
app = Celery(name='client', broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND)
# Send task to rabbitMq
task = app.send_task('task_1', args=("message produit",), retry=True)
# Wait the result into redis backend
result = task.get(CELERY_RESULT_TIMEOUT)
# Once the result is retrieved, forget the task inside redis backend to free memory
task.forget()
print (result)['task_1 message : message qui sera traité par task_1 uniquement', 'task_2 message : message qui sera traité par task_2 uniquement']Comme on a déjà bien fait le tour de celery, pour aller plus loin, je vous invite vivement à lire leur documentation qui est très bien faite et illustrée : https://docs.celeryproject.org/en/stable/
Le mot de la fin
Celery est un bon investissement dans de nombreux cas de figures et il vous permettra de gérer sereinement la montée en charge de vos applications les plus gourmandes. Alors, pourquoi s’en priver ?
Si cet article vous a plu ou si vous avez des questions, n’hésitez pas à partager vos états d’âme dans les commentaires sont là pour ça.
Merci Christophe, pour cet article très bien construit!.
Bon article mais bourré de fautes…dommage.
Bonjour,
Merci pour votre retour.
J’en ai profité pour faire une passe de correction (fautes d’inattention principalement).
Si vous en voyez encore, n’hésitez pas à me dire où et je les corrigerai.
Article tres complet sur Celery.