Bei den jüngsten Testarbeiten von Askaban ist es notwendig, Online-Planungsszenarien in der Testumgebung für Stabilitätstests zu simulieren. Deshalb kehrte ich zu meinem alten Geschäft mit Python zurück und schrieb Skripte in Python, um ähnliche Online-Planungsszenarien zu erstellen. Während des Skript-Schreibprozesses stieß ich auf eine Anforderung: 10.000 Jobströme in der Testumgebung zu erstellen.
Die ursprüngliche Idee bestand darin, die Schnittstelle zum Erstellen von Jobs 10.000 Mal in einer Schleife unter einem Askaban-Projekt aufzurufen (jeder Flow enthält nur einen Job). Da Askaban selbst über keine Schnittstelle zum Hinzufügen/Löschen von Jobströmen verfügt, werden alle Änderungen, Hinzufügungen und Löschungen des Jobstroms tatsächlich durch erneutes Hochladen des Projekt-Zip-Pakets implementiert. Dementsprechend erfolgt die Erstellung der Job-Schnittstelle des Mammoth-Frontends Wird angepasst, integriert Mammoth tatsächlich den Inhalt des Zip-Pakets erneut und lädt das Zip-Paket dann erneut nach Askaban hoch. Der gesamte Vorgang kann in die folgenden Prozesse unterteilt werden: Dekomprimieren des Zip-Pakets, um den Inhalt der Zip-Datei zu erhalten Paket, ändern Sie den Dateiinhalt im ZIP-Paket und verpacken Sie das auf Azkaban hochgeladene ZIP-Paket neu. Daher enthält das ZIP-Paket mit zunehmender Anzahl der Zyklen mehr Inhalt und die einmalige Ausführung der Schnittstelle dauert länger. In der Praxis hat sich herausgestellt, dass der erste Aufruf der Schnittstelle weniger als 1 Sekunde dauert, und wenn der Zyklus 1.000 Mal beträgt, beträgt die Zeit, die für den einmaligen Aufruf der Schnittstelle benötigt wird, fast 3 Sekunden. Wenn Sie also mit einer Schleife von 10.000 Wiederholungen rechnen, um diese Szene zu erstellen, wird dies offensichtlich sehr viel Zeit in Anspruch nehmen.
In diesem Zusammenhang liegt es nahe, über die Verwendung eines Multiprozess-/Multithread-Ansatzes zur Lösung dieses Problems nachzudenken.
Wie wir alle wissen, kann das Betriebssystem mehrere Aufgaben gleichzeitig ausführen. Sie hören beispielsweise Musik, chatten über Sofortnachrichten, schreiben einen Blog usw. Die meisten heutigen CPUs sind Multi-Core-CPUs, aber selbst Single-Core-CPUs unterstützten in der Vergangenheit die parallele Ausführung mehrerer Aufgaben.
Das Prinzip des Multitaskings auf einer Single-Core-CPU: Das Betriebssystem führt jede Aufgabe abwechselnd aus. Lassen Sie zunächst Aufgabe 1 0,01 Sekunden lang ausführen, wechseln Sie dann 0,01 Sekunden lang zu Aufgabe 2, wechseln Sie dann 0,01 Sekunden lang zu Aufgabe 3 ... und so weiter. Da die Ausführungsgeschwindigkeit der CPU sehr hoch ist, hat der Benutzer das subjektive Gefühl, dass diese Aufgaben parallel ausgeführt werden.
Das Prinzip der Multi-Core-CPU, die Multitasking ausführt: Da in tatsächlichen Anwendungen die Anzahl der Aufgaben die Anzahl der CPU-Kerne oft weit übersteigt, plant das Betriebssystem diese Multitasking tatsächlich -Aufgaben nacheinander auf jedem Kern ausführen.
Für das Betriebssystem ist eine Anwendung ein Prozess. Wenn Sie beispielsweise einen Browser öffnen, ist es ein Prozess; wenn Sie einen Notepad öffnen, ist es ein Prozess. Jeder Prozess hat seine eigene eindeutige Prozessnummer. Sie teilen sich die Speicherressourcen des Systems. Ein Prozess ist die kleinste Einheit für das Betriebssystem zur Zuweisung von Ressourcen .
Für jeden Prozess, der gleichzeitig Video und Audio abspielen muss, müssen mindestens zwei „Unteraufgaben“ gleichzeitig ausgeführt werden. Diese Unteraufgaben werden innerhalb des Prozesses ausgeführt Threads. Fertig. Ein Thread ist die kleinste Ausführungseinheit. Ein Prozess kann mehrere Threads enthalten, die unabhängig voneinander sind und die Ressourcen des Prozesses gemeinsam nutzen.
Multiprocessing ist ein von Python bereitgestelltes plattformübergreifendes Multiprozessmodul, mit dem Sie ganz einfach Multiprozess-Programmierung schreiben können. Prozessprogramme können auf verschiedenen Plattformen (Unix/Linux, Windows) ausgeführt werden.
Das Folgende ist der Code für die Verwendung von Multiprocessing zum Schreiben eines Multiprozessprogramms:
#!/usr/bin/python# -*- coding: utf-8 -*author = 'zni.feng'import sys
reload (sys)
sys.setdefaultencoding('utf-8')from multiprocessing import Processimport osimport time#子进程fundef child_projcess_fun(name): print 'Child process %s with processId %s starts.' % (name, os.getpid())
time.sleep(3) print 'Child process %s with processId %s ends.' % (name, os.getpid())if name == "main": print 'Parent processId is: %s.' % os.getpid()
p = Process(target = child_projcess_fun, args=('zni',)) print 'Process starts'
p.start() #开始进程
p.join() #等待子进程结束后再继续往下执行
print 'Process ends.'Ausgabe des Programms:
Parent processId is: 11076. Process starts Child process zni with processId 11077 starts. Child process zni with processId 11077 ends. Process ends. [Finished in 3.1s]
Bestimmt In einigen Fällen möchten wir mehrere Unterprozesse stapelweise erstellen oder eine Obergrenze für die Anzahl der Unterprozesse festlegen, um eine unbegrenzte Nutzung von Systemressourcen zu vermeiden. Diese Arbeit kann über Pool (Prozesspool) durchgeführt werden. Das Folgende ist der Code für die Verwendung von Pool:
1 #!/usr/bin/python 2 # -*- coding: utf-8 -* 3 author = 'zni.feng' 4 import sys 5 reload (sys) 6 sys.setdefaultencoding('utf-8') 7 8 from multiprocessing import Pool 9 import os, time10 11 def child_process_test(name, sleep_time):12 print 'Child process %s with processId %s starts.' % (name, os.getpid())13 time.sleep(sleep_time)14 print 'Child process %s with processId %s ends.' % (name, os.getpid())15 16 if name == "main":17 print 'Parent processId is: %s.' % os.getpid()18 p = Pool() #进程池默认大小是cpu的核数19 #p = Pool(10) #生成一个容量为10的进程池，即最大同时执行10个子进程20 for i in range(5):21 p.apply_async(child_process_test, args=('zni_'+str(i), i+1,)) #p.apply_async向进程池提交目标请求22 23 print 'Child processes are running.'24 p.close()25 p.join() #用来等待进程池中的所有子进程结束再向下执行代码，必须在p.close()或者p.terminate()之后执行26 print 'All Processes end.'
Ausgabe des Programms:
Parent processId is: 5050. Child processes are running. Child process zni_0 with processId 5052 starts. Child process zni_1 with processId 5053 starts. Child process zni_2 with processId 5054 starts. Child process zni_3 with processId 5055 starts. Child process zni_0 with processId 5052 ends. Child process zni_4 with processId 5052 starts. Child process zni_1 with processId 5053 ends. Child process zni_2 with processId 5054 ends. Child process zni_3 with processId 5055 ends. Child process zni_4 with processId 5052 ends. All Processes end. [Finished in 6.2s]
close()-Methode und The Unterschied zwischen Terminate()-Methoden:
Schließen: Schließen Sie den Prozesspool, damit keine neuen Prozesse hinzugefügt werden können. Bereits ausgeführte Prozesse warten mit der weiteren Ausführung, bis sie beendet sind.
Beenden: Beenden Sie den Thread-Pool zwangsweise, und der ausgeführte Prozess wird ebenfalls zwangsweise beendet.
Das Multiprocessing-Modul von Python bietet eine Vielzahl von Kommunikationsmethoden zwischen Prozessen, wie z. B. Warteschlange, Pipe usw.
3.1 Warteschlange, Sperre
Warteschlange ist ein durch Multiprocessing bereitgestelltes Modul. Seine Datenstruktur ist die „FIFO-First-In-First-Out“-Warteschlange. Häufig verwendete Methoden sind: put(object) Enter the queue; get() dequeues; empty() bestimmt, ob die Warteschlange leer ist.
Sperre: Wenn mehrere Unterprozesse Schreibvorgänge in derselben Warteschlange ausführen, kann eine Sperre verwendet werden, um einen Unterprozess über die einzige Schreibberechtigung für die Warteschlange zu verfügen, um Konflikte bei gleichzeitigen Vorgängen zu vermeiden. und andere Unterprozesse müssen Warten Sie, bis die Sperre aufgehoben wird, bevor Sie wieder mit dem Schreibvorgang beginnen können.
Das Folgende ist der Code für die Verwendung von Queue für die Kommunikation zwischen Prozessen: Erstellen Sie zwei untergeordnete Prozesse im übergeordneten Prozess, um jeweils Lese- und Schreibvorgänge in der Warteschlange zu implementieren.
1 #!/usr/bin/python 2 # -*- coding: utf-8 -* 3 author = 'zni.feng' 4 import sys 5 reload (sys) 6 sys.setdefaultencoding('utf-8') 7 from multiprocessing import Process, Queue, Lock 8 import os, time, random 9 #写数据进程10 def write(q, lock, name):11 print 'Child Process %s starts' % name12 #获得锁13 lock.acquire()14 for value in ['A' , 'B', 'C']:15 print 'Put %s to queue...' % value16 q.put(value)17 time.sleep(random.random())18 #释放锁19 lock.release()20 print 'Child Process %s ends' % name21 22 #读数据进程23 def read(q, lock, name):24 print 'Child Process %s starts' % name25 while True: #持续地读取q中的数据26 value =q.get()27 print 'Get %s from queue.' % value28 print 'Child Process %s ends' % name29 30 if name == "main":31 #父进程创建queue，并共享给各个子进程32 q= Queue()33 #创建锁34 lock = Lock()35 #创建第一个“写”子进程36 pw = Process(target = write , args=(q, lock, 'WRITE', ))37 #创建“读”进程38 pr = Process(target = read, args=(q,lock, 'READ',))39 #启动子进程pw,写入：40 pw.start()41 #启动子进程pr，读取：42 pr.start()43 #等待pw结束：44 pw.join()45 #pr是个死循环，通过terminate杀死：46 pr.terminate()47 print 'Test finish.'
Die Ausgabe von Das Programm lautet:
Child Process WRITE starts Put A to queue... Child Process READ starts Get A from queue. Put B to queue... Get B from queue. Put C to queue... Get C from queue. Child Process WRITE ends Test finish. [Finished in 2.0s]
3.2 Pipe
Pipe是另一种进程间通信的方式，俗称“管道”。它由两端组成，一端往管道里写入数据，另一端从管道里读取数据。
下面就是使用Pipe通信的代码：
1 #!/usr/bin/python 2 # -*- coding: utf-8 -* 3 author = 'zni.feng' 4 import sys 5 reload (sys) 6 sys.setdefaultencoding('utf-8') 7 from multiprocessing import Process, Pipe 8 import os, time, random 9 10 #发送数据进程11 def send(child_pipe, name):12 print 'Child Process %s starts' % name13 child_pipe.send('This is Mr.Ni')14 child_pipe.close()15 time.sleep(random.random())16 print 'Child Process %s ends' % name17 18 #接收数据进程19 def recv(parent_pipe, name):20 print 'Child Process %s starts' % name21 print parent_pipe.recv()22 time.sleep(random.random())23 print 'Child Process %s ends' % name24 25 if name == "main":26 #创建管道27 parent,child = Pipe()28 #创建send进程29 ps = Process(target=send, args=(child, 'SEND'))30 #创建recv进程31 pr = Process(target=recv, args=(parent, 'RECEIVE'))32 #启动send进程33 ps.start()34 #等待send进程结束35 ps.join()36 #启动recv进程37 pr.start()38 #等待recv进程结束39 pr.join()40 print 'Test finish.'
程序的输出结果如下：
Child Process SEND starts Child Process SEND ends Child Process RECEIVE starts This is Mr.Ni Child Process RECEIVE ends Test finish. [Finished in 1.8s]
【相关推荐】
2. Python中推荐使用多进程而不是多线程？分享推荐使用多进程的原因
Das obige ist der detaillierte Inhalt vonBeispiele für Multiprozess und Multithreading in Python (1). Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
