[Python] Erstellen Sie eine asynchrone Taskausführungsumgebung + Überwachungsumgebung
was ist das
Es gibt Zeiten, in denen Sie Aufgaben in einer Warteschlange speichern und als Worker behandeln möchten. Computeraufgaben, Masseneinsatz von etwas usw. Bei der Erstellung eines solchen Mechanismus ist Sellerie für Python praktisch.
Tor
Die Person, die dies liest, kann den Mechanismus der asynchronen Verarbeitung mit Python leicht erkennen und sogar den Verarbeitungsstatus überwachen. Was du tun kannst.
Vorbereitung
- Redis Server (flüchtige KVS läuft im Speicher)
- Sellerie (Bereitstellung eines Mechanismus für die asynchrone Verarbeitung)
- Blume (die, die durch die Ausdehnung von Sellerie überwacht wird)
apt-get install redis-server
pip install celery
pip install flower
Minimale Komponenten
main.py
import tasks
print('<first task>')
#Aufgabe hier starten(Aufgabe ausführen)
worker = tasks.run.delay()
#Wenn es nicht endet, warten Sie, bis es vorbei ist
while not worker.ready():
pass
#Geben Sie einen Rückgabewert an
print worker.result
print('<second task>')
#Aufgabe hier starten(calc Aufgabe)
worker = tasks.calc.delay(100, 200)
#Wenn es nicht endet, warten Sie, bis es vorbei ist
while not worker.ready():
pass
#Geben Sie einen Rückgabewert an
print worker.result
tasks.py Wenn Sie die Aufgaben, die Sie asynchron verarbeiten möchten, zu einer Funktion zusammenfügen und einen @ Task-Dekorator hinzufügen Bereit, aus Sellerie zu schlagen. Der Serializer von Sellerie handhabt die Übergabe von Argumenten und Rückgabewerten gut. Beachten Sie, dass Instanzen Ihrer eigenen Klasse nicht serialisiert werden können.
import time
from celery.decorators import task
@task
def run():
time.sleep(10)
print('Owata verarbeiten')
return 'ich bin fertig'
@task
def calc(a, b):
return a+b
celeryconfig.py Eine Einstellungsdatei zum Ausführen von Sellerie. Grundsätzlich Datenübertragung um Arbeitnehmer Ich möchte es mit json machen, also habe ich "json" als Serializer für die Aufgaben- / Ergebnislieferung angegeben. Das Backend (BROKER) ist für die Arbeit mit Redis ausgelegt, es kann jedoch auch Rabbit MQ verwendet werden. (Ich überlasse es dir) Im folgenden Beispiel lädt der Worker task.py. Funktionen, die asynchron verarbeitet werden sollen Geben Sie alle Skripte an, die eingeschlossen werden sollen. Wenn CELERYD_LOG_LEVEL auf INFO gesetzt ist, wird auch die Standardausgabe der Aufgabe protokolliert (celeryd.log). Geschrieben in. In der Produktion ist es möglicherweise besser, ERROR einzustellen.
- Übrigens wird gemunkelt, dass CELERYD_LOG_LEVEL dupliziert wurde.
Da CELERYD_CONCURRENCY = 1 ist, werden die Warteschlangen einzeln behandelt. Hier ist es besser, die Anzahl der CPUs anzupassen.
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", )
Wie man sich bewegt
Starten Sie den Redis-Server
Beginnen wir zuerst mit dem Redis-Server. (Erforderlich) Überspringen Sie diejenigen, die den Dienst bereits gestartet haben.
$ redis-server
Starten Sie den Selleriearbeiter
Der Worker ist jetzt bereit, die Warteschlange zu bearbeiten
(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
Stellen Sie tatsächlich eine Warteschlange
Asynchrone Verarbeitung
docker@1824542bb286:~/workspace$ python main.py
<first task>
ich bin fertig
<second task>
300
Aufgabenüberwachung durch Blume
Blumenstart
(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
Greifen Sie auf die Schnittstelle zu
Standardmäßig ist localhost: 555 die URL der Blume (Überwachungsschnittstelle). Dies ist praktisch, da Sie die Anzahl der Mitarbeiter sowie die Überwachung anpassen können.
Recommended Posts