Installez et utilisez le package de file d'attente de tâches asynchrone Celery dans l'environnement Python

1. Introduction

le céleri (céleri) est une file d'attente de tâches/file d'attente de travaux asynchrone basée sur la messagerie distribuée. Il se concentre sur les opérations en temps réel, mais dispose également d'un bon support de planification.
Le céleri est utilisé dans les systèmes de production pour gérer des millions de tâches chaque jour.
le céleri est écrit en Python, mais le protocole peut être implémenté dans n'importe quel langage. Il peut également être implémenté avec d’autres langages via des webhooks.
Le courtier de messages recommandé est RabbitMQ, mais offre une prise en charge limitée de Redis, Beanstalk, MongoDB, CouchDB et des bases de données (en utilisant SQLAlchemy ou Django ORM).
Celery est facile à intégrer à Django, Pylons et Flask, utilisez simplement les packages complémentaires django-celery, celery-pylons et Flask-Celery.

2. Installation
Avec les concepts ci-dessus, vous devez installer plusieurs choses : RabbitMQ, SQLAlchemy, Celery
La méthode d'installation est également très simple : RabbitMQ :
Mac :

brew install rabbitmq

linux :

sudo apt-get install rabbitmq-server

Les deux autres sont tous des éléments Python. Installez-les simplement directement avec pip. Les étudiants qui n'ont jamais installé le pilote MySQL devront peut-être installer MySQL-python.
Une fois l'installation terminée, démarrez le service :

$ rabbitmq-server[回车]

Ne fermez pas la fenêtre après le démarrage, effectuez les opérations suivantes pour créer une nouvelle fenêtre (onglet)

3. Cas simple
Assurez-vous que votre précédent RabbitMQ a été démarré.
Toujours l'exemple sur le site officiel, créez un fichier tâches.py dans n'importe quel répertoire avec le contenu suivant :

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def add(x, y):
  return x + y

à le même niveau d'exécution du répertoire :

$ celery -A tasks worker --loglevel=info

Cette commande signifie démarrer un travailleur et ajouter les tâches (add(x,y)) dans tâches La tâche est placée dans la file d'attente.
Gardez la fenêtre ouverte, ouvrez une nouvelle fenêtre pour passer en mode interactif, python ou ipython :

>>> from tasks import add
>>> add.delay(4, 4)

Jusqu'à présent, vous pouvez Utilisez du céleri pour exécuter la tâche. La tâche d'ajout est simplement appelée dans le mode interactif python ci-dessus et les paramètres 4, 4 sont transmis.
Mais il y a un problème en ce moment. Vous voulez soudainement connaître les résultats d'exécution et l'état de cette tâche, et si elle est terminée ou non. Par conséquent, vous devez configurer un backend.
Modifiez le code dans Tasks.py avant :

# coding:utf-8
import subprocess
from time import sleep

from celery import Celery

backend = 'db+mysql://root:@192.168.0.102/celery'
broker = 'amqp://guest@192.168.0.102:5672'

app = Celery('tasks', backend=backend, broker=broker)


@app.task
def add(x, y):
  sleep(10)
  return x + y


@app.task
def hostname():
  return subprocess.check_output(['hostname'])

En plus d'ajouter un backend, un who est également ajouté ci-dessus La méthode est utilisé pour tester le fonctionnement multi-serveur. Une fois la modification terminée, elle démarrera toujours comme avant.
Entrez de la même manière le modèle d'interaction python :

>>> from tasks import add, hostname
>>> r = add.delay(4, 4)
>>> r.ready() # 10s内执行，会输出False，因为add中sleep了10s
>>>
>>> r = hostname.delay()
>>> r.result # 输出你的hostname

4 Le test de plusieurs serveurs
est effectué. Après le test ci-dessus, un doute est apparu. Le céleri est appelé gestion des tâches distribuées, alors où est sa distribution ? Comment ses missions sont-elles accomplies ? Sur quelle machine a-t-il été exécuté ?
Sans arrêter le service Celery sur le serveur actuel, installez Celery sur un autre serveur de la même manière et démarrez-le :

$ celery -A tasks worker --loglevel=info

On constate que le service Celery du serveur précédent affiche le nom d'hôte du serveur que vous venez de démarrer, à condition que le serveur soit connecté à votre lapin mq.
Entrez ensuite dans le mode interactif python :

>>> from tasks import hostname
>>>
>>> for i in range(10):
...   r = hostname.delay()
...   print r.result # 输出你的hostname
>>>

Regardez ce que vous avez entré et observez comment vous avez démarré le service de céleri sur les deux sortie des serveurs.

5. Problèmes avec la connexion à distance RabbitMQ
Lors du test initial, le serveur distant n'a pas pu se connecter au service RabbitMQ local. Plus tard, il a été découvert que des autorisations devaient être définies. dans /usr/local/etc/rabbitmq Dans le fichier /rabbitmq-env.conf, modifiez l'ip dans NODE_IP_ADDRESS=127.0.0.1 en 0.0.0.0.

6. Pour résumer
Cet article présente brièvement l'utilisation du céleri, en mettant toujours l'accent sur une utilisation distribuée. La partie inconfortable est que lors de l'expansion, vous devez redéployer le code (tasks.py) au lieu de partager les tâches directement. Peut-être que Celery utilise des tâches pour faire correspondre différents travailleurs ? Je n’y connais pas encore grand chose, j’en parlerai après l’avoir utilisé en profondeur.

Pour plus d'articles sur l'installation et l'utilisation du package de file d'attente de tâches asynchrone Celery dans l'environnement Python, veuillez faire attention au site Web PHP chinois !