Comment vérifier l'état des tâches dans Celery?

Comment vérifier si une tâche est en cours dans le céleri (en particulier, j'utilise céleri-django)?

J'ai lu la documentation et j'ai cherché sur Google, mais je ne vois pas un appel comme:

my_example_task.state() == RUNNING

Mon cas d'utilisation est que j'ai un service externe (java) pour le transcodage. Lorsque j'envoie un document à transcoder, je souhaite vérifier si la tâche qui exécute ce service est en cours d'exécution, et sinon, la (re) démarrer.

J'utilise les versions stables actuelles - 2.4, je crois.

Renvoyez le task_id (qui est donné par .delay ()) et demandez ensuite à l'instance de céleri l'état:

x = method.delay(1,2)
print x.task_id

Lorsque vous le demandez, obtenez un nouveau AsyncResult en utilisant ce task_id:

from celery.result import AsyncResult
res = AsyncResult("your-task-id")
res.ready()

La création d'un AsyncResultobjet à partir de l'ID de tâche est la méthode recommandée dans la FAQ pour obtenir l'état de la tâche lorsque la seule chose dont vous disposez est l'ID de la tâche.

Cependant, à partir de Celery 3.x, il y a des mises en garde importantes qui pourraient mordre les gens s'ils ne font pas attention à eux. Cela dépend vraiment du scénario d'utilisation spécifique.

Par défaut, Celery n'enregistre pas un état "en cours d'exécution".

Pour que Celery enregistre qu'une tâche est en cours d'exécution, vous devez définir task_track_startedsur True. Voici une tâche simple qui teste ceci:

@app.task(bind=True)
def test(self):
    print self.AsyncResult(self.request.id).state

Quand task_track_startedest False, qui est la valeur par défaut, le show d'état est PENDINGmême si la tâche a commencé. Si vous définissez task_track_startedsur True, l'état sera STARTED.

L'État PENDINGsignifie «je ne sais pas».

Un AsyncResultavec l'étatPENDING ne veut rien dire de plus que le céleri ne connaît pas l'état de la tâche. Cela peut être dû à un certain nombre de raisons.

D'une part, AsyncResultpeut être construit avec des identifiants de tâches non valides. Ces «tâches» seront considérées comme en attente par Celery:

>>> task.AsyncResult("invalid").status
'PENDING'

Ok, donc personne ne va fournir des identifiants manifestement invalides AsyncResult. Assez juste, mais cela a aussi pour effet de AsyncResultconsidérer également une tâche qui s'est déroulée avec succès mais que Céleri a oublié comme étant PENDING. Encore une fois, dans certains scénarios de cas d'utilisation, cela peut être un problème. Une partie du problème dépend de la façon dont Celery est configuré pour conserver les résultats des tâches, car cela dépend de la disponibilité des «tombstones» dans le backend des résultats. ("Tombstones" est le terme utilisé dans la documentation Celery pour les blocs de données qui enregistrent la fin de la tâche.) L'utilisation AsyncResultne fonctionnera pas du tout si task_ignore_resultc'est le cas True. Un problème plus épineux est que Celery expire les pierres tombales par défaut. leresult_expiresle réglage par défaut est défini sur 24 heures. Donc, si vous lancez une tâche et enregistrez l'identifiant dans le stockage à long terme, et plus encore 24 heures plus tard, vous créez un AsyncResultavec elle, le statut sera PENDING.

Toutes les «vraies tâches» commencent dans l' PENDINGétat. Ainsi, entreprendre PENDINGune tâche pourrait signifier que la tâche a été demandée mais n'a jamais progressé plus loin que cela (pour une raison quelconque). Ou cela pourrait signifier que la tâche s'est exécutée mais que Celery a oublié son état.

Aie! AsyncResultne fonctionnera pas pour moi. Que puis-je faire d'autre?

Je préfère garder une trace des objectifs que de suivre les tâches elles-mêmes . Je garde certaines informations sur les tâches, mais elles sont vraiment secondaires au suivi des objectifs. Les buts sont stockés dans un stockage indépendant du céleri. Lorsqu'une demande doit effectuer un calcul dépend de l'atteinte d'un objectif, elle vérifie si l'objectif a déjà été atteint, si oui, elle utilise cet objectif mis en cache, sinon elle démarre la tâche qui affectera l'objectif, et envoie à le client qui a fait la demande HTTP une réponse qui indique qu'il doit attendre un résultat.

Les noms de variables et les hyperliens ci-dessus concernent Celery 4.x. Dans 3.x les variables et les liens hypertextes correspondants sont: CELERY_TRACK_STARTED, CELERY_IGNORE_RESULT, CELERY_TASK_RESULT_EXPIRES.

Chaque Taskobjet a une .requestpropriété, qui le contient AsyncRequestobjet. En conséquence, la ligne suivante donne l'état d'une tâche task:

task.AsyncResult(task.request.id).state

Vous pouvez également créer des états personnalisés et mettre à jour son exécution des tâches de valeur. Cet exemple est tiré de la documentation:

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

http://celery.readthedocs.org/en/latest/userguide/tasks.html#custom-states

Ancienne question mais j'ai récemment rencontré ce problème.

Si vous essayez d'obtenir le task_id, vous pouvez le faire comme ceci:

import celery
from celery_app import add
from celery import uuid

task_id = uuid()
result = add.apply_async((2, 2), task_id=task_id)

Vous savez maintenant exactement ce qu'est le task_id et pouvez maintenant l'utiliser pour obtenir le AsyncResult:

# grab the AsyncResult 
result = celery.result.AsyncResult(task_id)

# print the task id
print result.task_id
09dad9cf-c9fa-4aee-933f-ff54dae39bdf

# print the AsyncResult's status
print result.status
SUCCESS

# print the result returned 
print result.result
4

Utilisez simplement cette API de la FAQ sur le céleri

result = app.AsyncResult(task_id)

Cela fonctionne très bien.

Réponse de 2020:

#### tasks.py
@celery.task()
def mytask(arg1):
    print(arg1)

#### blueprint.py
@bp.route("/args/arg1=<arg1>")
def sleeper(arg1):
    process = mytask.apply_async(args=(arg1,)) #mytask.delay(arg1)
    state = process.state
    return f"Thanks for your patience, your job {process.task_id} \
             is being processed. Status {state}"

Essayer:

task.AsyncResult(task.request.id).state

cela fournira le statut de la tâche céleri. Si la tâche Celery est déjà en état d' échec , elle lancera une exception:

raised unexpected: KeyError('exc_type',)

pour des tâches simples, nous pouvons utiliser http://flower.readthedocs.io/en/latest/screenshots.html et http://policystat.github.io/jobtastic/ pour effectuer la surveillance.

et pour les tâches compliquées, disons une tâche qui traite de beaucoup d'autres modules. Nous vous recommandons d'enregistrer manuellement la progression et le message sur l'unité de tâche spécifique.

J'ai trouvé des informations utiles dans le

Guide des travailleurs du projet céleri inspecteurs-travailleurs

Pour mon cas, je vérifie si Celery fonctionne.

inspect_workers = task.app.control.inspect()
if inspect_workers.registered() is None:
    state = 'FAILURE'
else:
    state = str(task.state) 

Vous pouvez jouer avec inspect pour obtenir vos besoins.

• D'abord ， dans votre application céleri ：

vi my_celery_apps / app1.py

app = Celery(worker_name)

• et ensuite, passez au fichier de tâches ， importez l'application à partir de votre module d'application céleri.

vi tâches / task1.py

from my_celery_apps.app1 import app

app.AsyncResult(taskid)

try:
   if task.state.lower() != "success":
        return
except:
    """ do something """

En dehors de l'approche programmatique ci-dessus L'utilisation du statut de la tâche de fleur peut être facilement vue.

Surveillance en temps réel à l'aide des événements Celery. Flower est un outil Web permettant de surveiller et d'administrer les grappes de céleri.

1. Progression et historique des tâches
2. Possibilité d'afficher les détails de la tâche (arguments, heure de début, exécution, etc.)
3. Graphiques et statistiques

Document officiel: Fleur - Outil de surveillance du céleri

Installation:

$ pip install flower

Usage:

http://localhost:5555

res = method.delay()
    
print(f"id={res.id}, state={res.state}, status={res.status} ")

print(res.get())