3 cas pour utiliser Celery dans une application Django

Je crée des applications web sur Django. Fondamentalement, ce sont des services SaaS pour les entreprises. Toutes ces applications nécessitent des tâches asynchrones. Pour leur mise en œuvre, j'utilise du céleri. Dans l'article, je parlerai des situations dans lesquelles j'utilise Celery, avec des exemples de code.

Le céleri est un système de gestion des files d'attente de tâches. Fondamentalement capable 2 choses: prendre des tâches de la file d'attente et effectuer des tâches selon un calendrier. Le courtier de file d'attente est généralement RabbitMQ ou Redis. Les tâches sont placées dans la file d'attente, puis les employés de Celery les prennent à partir de là et les exécutent.

Pour le céleri, vous pouvez penser à une application dans presque toutes les applications, mais je décrirai ensuite uniquement les cas dans lesquels je l'utilise moi-même.

1. Tâches planifiées

Il y a souvent des tâches qui doivent être accomplies à une date et une heure spécifiques: envoyer un rappel à l'utilisateur, mettre fin à la période d'essai du compte, publier une publication sur les réseaux sociaux.

Dans Celery, il est possible de spécifier le paramètre ETA lors de l'appel de la tâche - l'heure à laquelle la tâche doit être lancée. Mais si vous planifiez des tâches de cette façon, cela se révèle très peu fiable: elles peuvent ne pas démarrer et sont mal à l'aise pour annuler.

Un moyen plus fiable consiste à utiliser le calendrier du céleri. C'est-à-dire, créez un calendrier où il y aura des tâches qui commencent à une certaine fréquence ou à une heure spécifique. Par exemple, si vous devez publier un article sur les réseaux sociaux selon un calendrier, la tâche est lancée une fois par minute. Si vous devez terminer la période d'essai pour votre compte, vous pouvez exécuter la tâche une fois par jour.

# schedule.py from datetime import timedelta from celery.schedules import crontab CELERYBEAT_SCHEDULE = { 'publish_post_starter': { 'task': 'publish_post_starter', 'schedule': timedelta(minutes=1), }, 'end_trial_starter': { 'task': 'end_trial_starter', 'schedule': crontab(hour=10, minute=21), }, } 

Dans le démarreur de tâches, nous obtenons toutes les instances pour lesquelles l'heure prévue est déjà arrivée. Nous passons par les instances et pour chacune, nous appelons la tâche principale. Comme arguments, nous transmettons uniquement l'ID d'instance afin de ne pas obstruer la file d'attente avec des données inutiles. Nous pouvons immédiatement parcourir toutes les instances et effectuer des actions, mais le plus souvent, il est préférable d'appeler une tâche distincte pour chaque instance. Nous allons donc accélérer l'exécution, et si une erreur se produit, cela n'affectera qu'une seule des tâches.

 # tasks.py @app.task(name='publish_post') def publish_post(post_id): ... @app.task(name='publish_post_starter') def publish_post_starter(): post_ids = list( Post.objects.filter( publish_dt__lte=timezone.now(), is_published=False ).values_list('id', flat=True) ) for post_id in post_ids: publish_post.delay(post_id) 

2. Appels informatiques et API longs de WSGI

WSGI fait référence au contexte dans lequel les demandes des utilisateurs sont traitées (cycle de demande-réponse). Contrairement au contexte des tâches asynchrones - Céleri.

Pour créer une interface réactive, tous les boutons doivent répondre instantanément et ne doivent pas bloquer le reste de l'interface. Pour ce faire, après avoir appuyé sur le bouton est bloqué, un spinner est placé dessus et une requête ajax est envoyée au serveur. Si le traitement de la demande prend plus de quelques secondes, vous pouvez déplacer le calcul vers la tâche Céleri.

Dans WSGI, nous appelons task et renvoyons une réponse. À l'avant, déverrouillez le bouton et retirez le cône. Nous montrons à l'utilisateur un message indiquant que l'action est en cours d'exécution. En parallèle, une tâche Celery est exécutée, qui, une fois terminée, renvoie une réponse sur le socket Web. Après avoir reçu le résultat à l'avant, nous le montrons à l'utilisateur.

 # rest_views.py from rest_framework import status from rest_framework.views import APIView from rest_framework.response import Response from tasks import send_emails class SendEmailView(APIView): def post(self, request): # this id will be used to send response with websocket request_uuid = request.data.get('request_uuid') if not request_uuid: return Response(status=status.HTTP_400_BAD_REQUEST) send_emails.delay(request.user.id, request_uuid) return Response(status=status.HTTP_200_OK) 

Séparément, vous pouvez distinguer les appels d'API externes de WSGI. Dans ce cas, tous les appels, quelle que soit la durée de leur exécution, sont lancés via la tâche Céleri. Ceci est une protection contre le fou. Il ne devrait pas y avoir de situation où, en raison de l'inaccessibilité de certaines API externes, l'interface utilisateur se bloque.

3. Défis de la tornade

Lors de l'intégration avec un réseau social, Telegram ou un service de paiement, vous avez besoin d'une URL de webhook à laquelle les notifications seront envoyées. Le nombre de demandes ne peut pas toujours être calculé à l'avance, mais leur nombre dépassera probablement les demandes des utilisateurs. Ces demandes seront reçues jusqu'à ce qu'elles reçoivent une réponse avec le code 200.

Pour le traitement de telles demandes, le cadre asynchrone Tornado convient. Afin de ne pas transformer le traitement en synchrone dans Tornado, il ne devrait pas y avoir d'opérations de blocage. C'est là que le céleri est nécessaire. Le gestionnaire Tornado reçoit la demande, valide les données, appelle la tâche Celery et renvoie une réponse réussie.

 # tornado_handlers.py from tornado import gen, escape from tornado.web import RequestHandler from tasks import handle_vk_callback class VkCallbackHandler(RequestHandler): @gen.coroutine def post(self, *args, **kwargs): try: data = escape.json_decode(self.request.body) except ValueError: self.set_status(status_code=400, reason='Invalid data') return handle_vk_callback.delay(data) self.write('ok') return