Cet article explore la mise en place d'un système de files d'attente de tâches asynchrones en utilisant Python 3.7, le framework web Tornado 5.1.1, la bibliothèque de gestion de tâches Celery 3.1 et le courtier de messages RabbitMQ 3.7.16. Ce processus permettra de déléguer des opérations potentiellement longues, comme l'envoi d'e-mails, à des procsesus en arrière-plan, améliorant ainsi la réactivité de l'application principale. Nous intégrerons également Flower pour la surveillance des files d'attente.

1. Installation de RabbitMQ

RabbitMQ est un courtier de messages robuste et performant, idéal pour Celery.

Sur macOS

Utilisez Homebrew pour une installation rapide :

# Installation du service RabbitMQ
brew install rabbitmq
# Démarrage du service
brew services start rabbitmq

Sur Windows 10

RabbitMQ étant basé sur Erlang, vous devez d'abord installer Erlang.

1. Téléchargez et installez Erlang pour Windows depuis le site officiel erlang.org/downloads (utilisez le répertoire d'installation par défaut).
2. Téléchargez et installez RabbitMQ depuis le site officiel rabbitmq.com/download.html (utilisez le répertoire d'installation par défaut).

Après l'installation, activez l'interface de gestion web de RabbitMQ. Naviguez vers le répertoire sbin de votre installation RabbitMQ (par exemple, C:\Program Files\RabbitMQ Server\rabbitmq_server-3.x.x\sbin) et exécutez la commande suivante :

rabbitmq-plugins enable rabbitmq_management

Démarrez le service RabbitMQ via le menu Démarrer de Windows. Ensuite, accédez à http://localhost:15672/ dans votre navigateur. Utilisez les identifiants par défaut guest/guest pour vous connecter au panneau de contrôle. Une connexion réussie confirme que RabbitMQ est opérationnel.

2. Installation des dépendances Python

Installez les bibliothèques Python nécessaires avec pip3, en spécifiant les versions pour correspondre à l'environnement cible :

pip3 install tornado==5.1.1
pip3 install celery==3.1
pip3 install pika==0.9.14
pip3 install tornado-celery
pip3 install flower

Note sur la compatibilité Python 3.7+ : Il est important de noter qu'avec Python 3.7+, le mot-clé async a été introduit, ce qui peut créer des conflits avec certaines variables ou fonctions nommées async dans des versions plus anciennes de bibliothèques tierces comme pika ou celery 3.1. Historiquement, pour utiliser ces versions spécifiques avec Python 3.7, certains développeurs ont rencontré le besoin d'effectuer des modifications manuelles dans le code source de ces bibliothèques (par exemple, renommer async en async_my). Cette approche n'est généralement pas recommandée pour la stabilité à long terme. Il est préférable d'utiliser des versions de Celery et Pika entièrement compatibles avec votre version de Python, ou d'envisager un environnement Python plus ancien si ces versions spécifiques sont impératives.

3. Définition des tâches Celery

Créez un fichier nommé taches_asynchrones.py dans le répertoire de votre projet. Ce fichier contiendra les définitions des tâches Celery.

import time
from celery import Celery

# Une fonction utilitaire fictive pour simuler l'envoi d'e-mail
def envoyer_courriel_simule(sujet, corps, destinataire):
    print(f"Simulation d'envoi d'e-mail à {destinataire} - Sujet: {sujet}, Corps: {corps}")
    time.sleep(2) # Simule un délai réseau ou de traitement
    print(f"E-mail envoyé à {destinataire}")
    return True

# Permet l'exécution en tant que root si nécessaire (pour certains environnements)
FORCE_EXECUTION_RACINE = True

# Configuration de l'application Celery
app_celery = Celery("mon_projet_taches", broker="amqp://guest:guest@localhost:5672")
app_celery.conf.CELERY_RESULT_BACKEND = "amqp"

@app_celery.task
def simuler_traitement_long(duree_secondes):
    """Simule une tâche longue en attendant un certain nombre de secondes."""
    time.sleep(float(duree_secondes))
    return f"Traitement terminé après {duree_secondes} secondes."

@app_celery.task
def expedier_courriel(sujet_mail, contenu_mail, adresse_destinataire):
    """Tâche Celery pour l'envoi d'e-mails."""
    succes = envoyer_courriel_simule(sujet_mail, contenu_mail, adresse_destinataire)
    if succes:
        return f"Courriel à {adresse_destinataire} envoyé avec succès."
    else:
        return f"Échec de l'envoi du courriel à {adresse_destinataire}."

if __name__ == "__main__":
    app_celery.start()

4. Implémentation du gestionnaire Tornado

Dans votre module de vues (par exemple, vues/principale.py), créez un gestionnaire Tornado qui déclenchera une tâche Celery de manière asynchrone.

import sys
import tcelery
from tornado import gen, web

# Assurez-vous que le chemin vers taches_asynchrones.py est correct
sys.path.append("..") # ou le chemin approprié
import taches_asynchrones as app_taches_celery

# Classe de base pour les gestionnaires (peut être définie ailleurs, ex: base_handlers.py)
class GestionnaireBase(web.RequestHandler):
    pass

# Gestionnaire Tornado pour déclencher l'envoi d'e-mail asynchrone
class GestionnaireEnvoiCourriel(GestionnaireBase):
    @gen.coroutine
    def get(self):
        # Déclenche la tâche d'envoi de courriel de manière asynchrone
        # Utilisation de apply_async avec gen.Task pour Tornado 5.x
        resultat_tache = yield gen.Task(
            app_taches_celery.expedier_courriel.apply_async,
            args=['Bienvenue !', 'Votre compte est activé.', 'utilisateur@example.com']
        )
        self.write(f'Tâche d\'envoi d\'e-mail soumise: {resultat_tache.id}')
        self.finish()

5. Configuration du routeur Tornado

Définissez les routes de votre application Tornado. Créez un fichier application_config.py :

import tornado.web
from vues import principale as vues_app
import config

# Une configuration simple (par exemple, dans config.py)
class AppConfig:
    parametres = {
        "debug": True
    }
    port_ecoute = 8000

# Routeur de l'application
class ConfigurationApplication(tornado.web.Application):
    def __init__(self):
        routes = [
            (r"/envoyer-courriel", vues_app.GestionnaireEnvoiCourriel)
        ]
        super(ConfigurationApplication, self).__init__(routes, **config.AppConfig.parametres)

6. Point d'entrée du serveur Tornado

Le fichier principal de démarrage du serveur, par exemple demarrer_serveur.py :

import tornado.ioloop
import tornado.httpserver
from application_config import ConfigurationApplication
from config import AppConfig # Assurez-vous d'importer votre configuration

if __name__ == "__main__":
    print('Démarrage du serveur Tornado...')
    instance_app = ConfigurationApplication()
    serveur_http = tornado.httpserver.HTTPServer(instance_app)
    
    # Liaison au port spécifié dans la configuration
    serveur_http.bind(AppConfig.port_ecoute)
    
    # Démarrage d'un seul processus worker pour le serveur Tornado
    # Pour la production, on pourrait utiliser plusieurs processus (start(0) pour CPU cores)
    serveur_http.start(1) 
    
    print(f"Serveur Tornado écoutant sur le port {AppConfig.port_ecoute}")
    tornado.ioloop.IOLoop.current().start()

7. Lancement des services

Ouvrez trois terminaux distincts et exécutez les commandes suivantes depuis le répertoire racine de votre projet :

Démarrer le serveur Tornado

python demarrer_serveur.py

Démarrer un worker Celery

celery -A taches_asynchrones worker --loglevel=info

Démarrer l'interface de surveillance Flower

celery flower -A taches_asynchrones --broker=amqp://guest:guest@localhost:5672//

8. Test et surveillance

Une fois tous les services lancés :

1. Ouvrez votre navigateur et accédez à l'URL suivante pour déclencher la tâche d'envoi d'e-mail : http://localhost:8000/envoyer-courriel.
2. Le terminal du worker Celery affichera le traitement et le résultat de la tâche.
3. Pour surveiller les tâches en temps réel, visitez l'interface Flower : http://localhost:5555/. Vous y verrez l'état de vos tâches (en cours, réussies, échouées).