Qu'est-ce que c'est ça

Il y a des moments où vous souhaitez stocker des tâches dans une file d'attente et les gérer en tant que travailleur. Tâches de calcul, déploiement en masse de quelque chose, etc. Lors de la création d'un tel mécanisme, céleri est pratique pour Python.

objectif

La personne qui lit ceci peut facilement réaliser le mécanisme de traitement asynchrone avec Python et même surveiller l'état du traitement. Ce que tu peux faire.

Préparation

serveur redis (KVS volatile fonctionnant en mémoire)
céleri (fournissant un mécanisme de traitement asynchrone)
fleur (celle qui est surveillée par l'extension du céleri)

apt-get install redis-server
pip install celery
pip install flower

Composants minimaux

main.py

import tasks

print('<first task>')
#Commencer la tâche ici(exécuter la tâche)
worker = tasks.run.delay()
#Si ça ne se termine pas, attendez que ce soit fini
while not worker.ready():
    pass
#Donnez une valeur de retour
print worker.result

print('<second task>')
#Commencez la tâche ici(tâche de calcul)
worker = tasks.calc.delay(100, 200)
#Si ça ne se termine pas, attendez que ce soit fini
while not worker.ready():
    pass
#Donnez une valeur de retour
print worker.result

tasks.py Si vous regroupez les tâches que vous souhaitez traiter de manière asynchrone dans une fonction et ajoutez un décorateur @task Prêt à frapper du céleri. Le sérialiseur de céleri gère bien le passage des arguments et les valeurs de retour. Notez que les instances de votre propre classe ne peuvent pas être sérialisées.

import time
from celery.decorators import task

@task
def run():
    time.sleep(10)
    print('Traitement Owata')
    return 'J'ai fini'


@task
def calc(a, b):
    return a+b

celeryconfig.py Un fichier de réglage pour courir le céleri. Fondamentalement, transfert de données autour des travailleurs Je veux le faire avec json, j'ai donc spécifié "json" comme sérialiseur pour la livraison des tâches / résultats. Le back-end (BROKER) est conçu pour fonctionner avec redis, mais Rabbit MQ peut également être utilisé. (Je te laisse ça) Dans l'exemple ci-dessous, le travailleur charge tasks.py. Fonctions à traiter de manière asynchrone Spécifions tous les scripts à inclure. Si CELERYD_LOG_LEVEL est défini sur INFO, la sortie standard de la tâche sera également enregistrée (celeryd.log). Écrit en. En production, il peut être préférable de le définir sur ERROR.

Au fait, la rumeur veut que CELERYD_LOG_LEVEL soit devenu dupliqué.

Puisque CELERYD_CONCURRENCY = 1, nous traiterons les files d'attente une par une. Il vaut mieux régler ici en fonction du nombre de processeurs.

BROKER_URL = 'redis://localhost/0'
CELERYD_CONCURRENCY = 1
CELERY_RESULT_BACKEND = 'redis'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_BACKEND = "redis"
CELERYD_LOG_FILE = "./celeryd.log"
CELERYD_LOG_LEVEL = "INFO"
CELERY_IMPORTS = ("tasks", )

Comment bouger

Démarrez redis-server

Commençons par démarrer redis-server. (Obligatoire) Ignorez ceux qui ont déjà démarré le service.

$ redis-server

Démarrer le céleri

Le worker est maintenant prêt à gérer la file d'attente

(env) docker@1824542bb286:~/workspace$ celery worker
/home/docker/.virtualenvs/env2/local/lib/python2.7/site-packages/celery/app/defaults.py:251: CPendingDeprecationWarning:
    The 'CELERYD_LOG_LEVEL' setting is scheduled for deprecation in     version 2.4 and removal in version v4.0.     Use the --loglevel argument instead

  alternative='Use the {0.alt} instead'.format(opt))
/home/docker/.virtualenvs/env2/local/lib/python2.7/site-packages/celery/app/defaults.py:251: CPendingDeprecationWarning:
    The 'CELERYD_LOG_FILE' setting is scheduled for deprecation in     version 2.4 and removal in version v4.0.     Use the --logfile argument instead

  alternative='Use the {0.alt} instead'.format(opt))

 -------------- celery@1824542bb286 v3.1.23 (Cipater)
---- **** -----
--- * ***  * -- Linux-3.13.0-24-generic-x86_64-with-Ubuntu-14.04-trusty
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         default:0x7f068383f610 (.default.Loader)
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:
- *** --- * --- .> concurrency: 1 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[tasks]
  . tasks.run

Je vais effectivement mettre une file d'attente

Traitement asynchrone

docker@1824542bb286:~/workspace$ python main.py
<first task>
J'ai fini
<second task>
300

Suivi des tâches par fleur

début de fleur

(env2) docker@1824542bb286:~/workspace$ celery flower
/home/docker/.virtualenvs/env2/local/lib/python2.7/site-packages/celery/app/defaults.py:251: CPendingDeprecationWarning:
    The 'CELERYD_LOG_LEVEL' setting is scheduled for deprecation in     version 2.4 and removal in version v4.0.     Use the --loglevel argument instead

  alternative='Use the {0.alt} instead'.format(opt))
/home/docker/.virtualenvs/env2/local/lib/python2.7/site-packages/celery/app/defaults.py:251: CPendingDeprecationWarning:
    The 'CELERYD_LOG_FILE' setting is scheduled for deprecation in     version 2.4 and removal in version v4.0.     Use the --logfile argument instead

  alternative='Use the {0.alt} instead'.format(opt))
[I 160617 13:02:20 command:136] Visit me at http://localhost:5555
[I 160617 13:02:20 command:141] Broker: redis://localhost:6379/0
[I 160617 13:02:20 command:144] Registered tasks:
    ['celery.backend_cleanup',
     'celery.chain',
     'celery.chord',
     'celery.chord_unlock',
     'celery.chunks',
     'celery.group',
     'celery.map',
     'celery.starmap',
     'tasks.run']
[I 160617 13:02:20 mixins:231] Connected to redis://localhost:6379/0

Accéder à l'interface

Par défaut, localhost: 555 est l'URL de la fleur (interface de surveillance). C'est pratique car vous pouvez ajuster le nombre de travailleurs ainsi que la surveillance.

[Python] Créer un environnement d'exécution de tâches asynchrone + environnement de surveillance