Flux d'obtention du résultat du traitement asynchrone à l'aide de Django et Celery

Aperçu

-Traitement asynchrone en combinant django et celery Flux jusqu'à l'enregistrement du résultat de l'utilisation de django-celery-results

• Un mémorandum de la pierre d'achoppement selon laquelle l'état reste PENDING ou le résultat n'est pas écrit dans le DB tant que le processus asynchrone n'est pas terminé.

supposition

--Environnement - python: v.3.7.7 - redis: v.6.0.9

• pip install
  • django: v.2.0.3
  • celery: v.4.4.7
  • django-redis: v.4.12.1
  • django-celery-results: v.1.2.1

Installation

--Modifier settings.py

INSTALLED_APPS = [
    ...
    'django_celery_results', #ajouter à
]

CELERY_BROKER_URL = "redis:// [redis host]:6379" #Spécifiez redis comme courtier
CELERY_RESULT_BACKEND = "django-db" #Le résultat est enregistré dans la base de données spécifiée par django. Cet article suppose MySQL.
CELERY_TASK_TRACK_STARTED = True #Paramètres de confirmation du démarrage de la tâche (décrits plus loin)

Construire

• Structure du fichier d'exécution
• La structure des fichiers créés ci-dessous est la suivante

├── myapp
│   ├── myapp
│   │   ├── __init__.py
│   │   ├── settings.py
│   │   ├── celery.py
│   │   ├── urls.py
│   │   ├── tasks
│   │   │   ├── __init__.py
│   │   │   └── task.py
│   │   └── wsgi.py
│   └── manage.py
├── docker-compose.yml
└── requirements.txt

--Créer celery.py --Créez le celery.py suivant sous le répertoire de l'application (même niveau que urls.py) (presque le même que le document officiel)

# celery.py
from __future__ import absolute_import, unicode_literals

import os

from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault(
    'DJANGO_SETTINGS_MODULE',
    'myapp.settings'
)

app = Celery('myapp') 

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object(
    'django.conf:settings',
    namespace='CELERY'
)

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

--Créer un fichier de tâche

• Afin de gérer plusieurs scripts asynchrones, le répertoire des tâches est coupé et les scripts des tâches sont placés en dessous. --Préparez task.py comme suit

##Appelé(=Task)
from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def hello(proj_id):
    time.sleep(10) #10 secondes pour confirmer que le processus se termine de manière asynchrone
    message = "hello"
    print(message) #Sortie standard
    return message #Renvoie le résultat

lancement

--Exécuter l'application Django

$python manage.py makemigrations #Créez un fichier de migration (si nécessaire)
$python manage.py migrate #exécuter migrer
$python manage.py rumserver #Exécution de l'application

• Lancement de Redis

# redis-server

• Lancement de bijoux --cd dans le répertoire où se trouve manage.py et exécutez la commande suivante

$celery -A myapp worker --concurrency=1 -l info

Exécuter la tâche pour obtenir l'ID

• L'exécution de la tâche peut être effectuée avec [your_task_function] .delay () --Dans l'exemple ci-dessous, il attend juste 10 secondes puis renvoie "bonjour".
• Afin de vérifier le contenu de la tâche, il est nécessaire d'obtenir l'ID attribué à la tâche.
• Un objet de tâche est retourné comme valeur de retour de hello.delay (), et l'identifiant peut être référencé.
• En conservant l'ID de tâche dans une base de données, etc., vous pouvez vérifier l'état de la tâche à tout moment dans l'application ou à partir de la logique d'un autre processus.

from django_app.tasks import hello
from celery.result import AsyncResult

def hoge():
    task = hello.delay(project_id) #C'est tout ce dont vous avez besoin pour appeler le traitement asynchrone. Après cela, le processus se déroule de manière asynchrone
    print("===============")
    print(task.id) #Vous pouvez vérifier l'ID attribué de manière unique.

Vérifiez l'état de la tâche

--Vérifiez l'état de la tâche à l'aide de l'ID de tâche obtenu dans l'exemple ci-dessus. --Utilisez AsyncResult pour vérifier l'état et le résultat

• Vous pouvez obtenir l'objet de tâche en transmettant l'ID de tâche.
• Vous pouvez obtenir le statut de la tâche avec la méthode status --PENDING: En attente d'exécution. --STARTED: état de démarrage de l'exécution. (Elle n'est pas sortie par défaut. La méthode de sortie sera décrite plus loin.) -- SUCCÈS: Exécution terminée normalement. --FAILED: l'exécution s'est terminée anormalement.

from django_app.tasks import hello
from celery.result import AsyncResult

def hoge():
    task = hello.delay(project_id) #C'est tout ce dont vous avez besoin pour appeler le traitement asynchrone. Après cela, le processus se déroule de manière asynchrone
    print("===============")
    task_id = task.id
    print(task_id) #Vous pouvez vérifier l'ID attribué de manière unique.
    task_1 = AsyncResult(task_id)
    print(task_1.status) #Vérifiez l'état immédiatement après le début du traitement
    time.sleep(1)
    print("===============")
    task_2 = AsyncResult(task_id)
    print(task_2.status) #Vérifiez l'état 1 seconde après le traitement

--Enregistrer le résultat dans DB «Je ne pense pas que vous l'utilisiez beaucoup, mais comme méthode. ..

• En spécifiant CELERY_RESULT_BACKEND dans settings.py, l'état / le résultat du traitement asynchrone est automatiquement enregistré dans la base de données. Dans cet article, nous supposerons MySQL.
• Vous pouvez rechercher des informations sur la tâche à partir de django en spécifiant l'ID de la tâche.

from django_app.tasks import hello
from django_celery_results.models import TaskResult

def hoge():
    task = hello.delay(project_id) #C'est tout ce dont vous avez besoin pour appeler le traitement asynchrone. Après cela, le processus se déroule de manière asynchrone
    print("===============")
    task_id = task.id #Vous pouvez vérifier l'ID attribué de manière unique.
    print(task_id) 
    task_model_1 = TaskResult.objects.filter(task_id=task_id)
    if len(task_model_1) > 0:
        print(task_model_1[0].status) #Vérifiez l'état immédiatement après le début du traitement
    time.sleep(1)
    task_model_2 = TaskResult.objects.filter(task_id=task_id)
    if len(task_model_2) > 0:
        print(task_model_2[0].status) #Vérifiez l'état 1 seconde après le traitement

#Contenu de MySQL

##Le tableau est le suivant.
mysql> show tables;
+----------------------------------+
| Tables_in_nodeai                 |
+----------------------------------+
| auth_group                       |
| auth_group_permissions           |
| auth_permission                  |
| django_admin_log                 |
| django_celery_results_taskresult |  ## <-Cela a été ajouté!
...

## django_celery_results_Le schéma de résultat de la tâche est le suivant.
mysql> desc django_celery_results_taskresult;
+------------------+--------------+------+-----+---------+----------------+
| Field            | Type         | Null | Key | Default | Extra          |
+------------------+--------------+------+-----+---------+----------------+
| id               | int(11)      | NO   | PRI | NULL    | auto_increment |
| task_id          | varchar(255) | NO   | UNI | NULL    |                |
| status           | varchar(50)  | NO   | MUL | NULL    |                |
| content_type     | varchar(128) | NO   |     | NULL    |                |
| content_encoding | varchar(64)  | NO   |     | NULL    |                |
| result           | longtext     | YES  |     | NULL    |                |
| date_done        | datetime(6)  | NO   | MUL | NULL    |                |
| traceback        | longtext     | YES  |     | NULL    |                |
| meta             | longtext     | YES  |     | NULL    |                |
| task_args        | longtext     | YES  |     | NULL    |                |
| task_kwargs      | longtext     | YES  |     | NULL    |                |
| task_name        | varchar(255) | YES  | MUL | NULL    |                |
| worker           | varchar(100) | YES  | MUL | NULL    |                |
| date_created     | datetime(6)  | NO   | MUL | NULL    |                |
+------------------+--------------+------+-----+---------+----------------+

Empêcher l'état de la tâche de rester PENDING

--Si CELERY_TASK_TRACK_STARTED n'est pas spécifié dans settings.py, la tâche sera exécutée jusqu'à ce que le résultat soit obtenu.

• Le statut obtenu en utilisant AsyncResult est toujours PENDING --Le contenu de la tâche n'est pas conservé dans le DB. (Enregistré uniquement lorsque le résultat est disponible)
• Afin de le résoudre, les paramètres suivants sont placés dans settings.py

CELERY_TASK_TRACK_STARTED = True

Essayez d'échouer la tâche

--Si vous soulevez une erreur avec relance, le résultat de la tâche sera FAILED

• Le message d'erreur résultant de la levée est stocké dans task.result

##Appelé(=Task)
from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def hello(proj_id):
    time.sleep(10) #10 secondes pour confirmer que le processus se termine de manière asynchrone
    message = "hello"
    raise Exception("my error message") #Notification d'erreur

#Résultats stockés dans MySQL
mysql> select * from django_celery_results_taskresult\G
...
*************************** 31. row ***************************
              id: 31
         task_id: be294008-d2fc-4760-9055-483efdaa4970
          status: FAILURE
    content_type: application/json
content_encoding: utf-8
          result: {"exc_type": "Exception", "exc_message": ["my error message"], "exc_module": "builtins"}
       date_done: 2020-11-10 08:06:32.848782
       traceback: Traceback (most recent call last):...
            meta: {"children": []}
       task_args: (4,)
     task_kwargs: {}
       task_name: myapp.tasks.task.hello
          worker: celery@05ab2e4b5ee1
    date_created: 2020-11-10 08:06:22.829301