Was ist die Kommunikationsmethode zwischen Python-Prozessen?

Einführung in die Put-Funktion: Daten übergeben. Es verfügt über eine Parameternachricht, bei der es sich um einen Zeichenfolgentyp handelt.

Einführung in die Get-Funktion: Wird zum Empfangen von Daten in der Warteschlange verwendet. (Eigentlich handelt es sich hierbei um ein häufiges JSON-Szenario. Ein großer Teil der Datenübertragung erfolgt in Zeichenfolgen. Das Einfügen und Abrufen von Warteschlangen erfolgt mithilfe von Zeichenfolgen, sodass JSON für dieses Szenario sehr gut geeignet ist.)

Als Nächstes üben wir die Verwendung von Warteschlangen.

Interprozesskommunikation – Warteschlangen-Demonstrationsfall

Das Codebeispiel lautet wie folgt:

# coding:utf-8


import json
import multiprocessing


class Work(object):     # 定义一个 Work 类
    def __init__(self, queue):      # 构造函数传入一个 '队列对象' --> queue
            self.queue = queue

    def send(self, message):        # 定义一个 send(发送) 函数，传入 message
                                    # [这里有个隐藏的bug，就是只判断了传入的是否字符串类型；如果传入的是函数、类、集合等依然会报错]
        if not isinstance(message, str):    # 判断传入的 message 是否为字符串，若不是，则进行 json 序列化
            message = json.dumps(message)
        self.queue.put(message)     # 利用 queue 的队列实例化对象将 message 发送出去

    def receive(self):      # 定义一个 receive(接收) 函数，不需传入参数，但是因为接收是一个源源不断的过程，所以需要使用 while 循环
        while 1:
            result = self.queue.get()   # 获取 '队列对象' --> queue 传入的message
                                        # 由于我们接收的 message 可能不是一个字符串，所以要进程异常的捕获
            try:                        # 如果传入的 message 符合 JSON 格式将赋值给 res ；若不符合，则直接使用 result 赋值 res
                res = json.loads(result)
            except:
                res = result
            print('接收到的信息为：{}'.format(res))


if __name__ == '__main__':
    queue = multiprocessing.Queue()
    work = Work(queue)
    send = multiprocessing.Process(target=work.send, args=({'message': '这是一条测试的消息'},))
    receive = multiprocessing.Process(target=work.receive)

    send.start()
    receive.start()

Bei Verwendung der Warteschlange zum Einrichten der Interprozesskommunikation ist eine Ausnahme aufgetreten

Hier wird jedoch ein Fehler angezeigt, wie unten gezeigt:

Ein Beispiel Der Fehler-Screenshot lautet wie folgt:

Die Fehlermeldung hier bedeutet, dass die Datei nicht gefunden wurde. Wenn wir die Warteschlange zum Ausführen von put() und get() verwenden, wird tatsächlich eine unsichtbare Sperre hinzugefügt, nämlich die .SemLock im Kreis im Bild oben. Wir müssen uns nicht um die konkrete Ursache dieses Fehlers kümmern. Die Lösung dieses Problems ist eigentlich sehr einfach.

FileNotFoundError: [Errno 2] Keine solche Datei- oder Verzeichnisausnahmelösung

Was den Prozess blockieren muss, ist nur einer der Sende- oder Empfangsunterprozesse. Blockieren Sie einfach einen davon. Aber unser Empfangsunterprozess ist eine While-Schleife, die immer ausgeführt wird, sodass wir nur einen Join zum Sendeunterprozess hinzufügen müssen.

Das Lösungsdiagramm sieht wie folgt aus:

PS: Obwohl das Fehlerproblem behoben wurde, wurde das Programm nicht normal beendet.

Da es sich bei unserem Empfangsprozess tatsächlich um eine While-Schleife handelt, wissen wir nicht, wann er verarbeitet wird, und es gibt keine Möglichkeit, ihn sofort zu beenden. Daher müssen wir im Empfangsprozess die Funktion „terminate()“ verwenden, um das empfangende Ende zu beenden.

Die laufenden Ergebnisse sind wie folgt:

Fügen Sie Daten stapelweise zur Sendefunktion hinzu

Erstellen Sie eine neue Funktion und schreiben Sie sie in eine for-Schleife, um das Hinzufügen von Nachrichten zu simulieren, die stapelweise gesendet werden sollen

Dann fügen Sie einen Thread hinzu zu dieser Funktion, die das Senden von Daten in Stapeln simuliert.

Der Beispielcode lautet wie folgt:

# coding:utf-8


import json
import time
import multiprocessing


class Work(object):     # 定义一个 Work 类
    def __init__(self, queue):      # 构造函数传入一个 '队列对象' --> queue
            self.queue = queue

    def send(self, message):        # 定义一个 send(发送) 函数，传入 message
                                    # [这里有个隐藏的bug，就是只判断了传入的是否字符串类型；如果传入的是函数、类、集合等依然会报错]
        if not isinstance(message, str):    # 判断传入的 message 是否为字符串，若不是，则进行 json 序列化
            message = json.dumps(message)
        self.queue.put(message)     # 利用 queue 的队列实例化对象将 message 发送出去


    def send_all(self):             # 定义一个 send_all(发送)函数，然后通过for循环模拟批量发送的 message
        for i in range(20):
            self.queue.put('第 {} 次循环，发送的消息为：{}'.format(i, i))
            time.sleep(1)



    def receive(self):      # 定义一个 receive(接收) 函数，不需传入参数，但是因为接收是一个源源不断的过程，所以需要使用 while 循环
        while 1:
            result = self.queue.get()   # 获取 '队列对象' --> queue 传入的message
                                        # 由于我们接收的 message 可能不是一个字符串，所以要进程异常的捕获
            try:                        # 如果传入的 message 符合 JSON 格式将赋值给 res ；若不符合，则直接使用 result 赋值 res
                res = json.loads(result)
            except:
                res = result
            print('接收到的信息为：{}'.format(res))


if __name__ == '__main__':
    queue = multiprocessing.Queue()
    work = Work(queue)
    send = multiprocessing.Process(target=work.send, args=({'message': '这是一条测试的消息'},))
    receive = multiprocessing.Process(target=work.receive)
    send_all = multiprocessing.Process(target=work.send_all,)


    send_all.start()    # 这里因为 send 只执行了1次，然后就结束了。而 send_all 却要循环20次，它的执行时间是最长的，信息也是发送的最多的
    send.start()
    receive.start()

    # send.join()       # 使用 send 的阻塞会造成 send_all 循环还未结束 ，receive.terminate() 函数接收端就会终结。
    send_all.join()     # 所以我们只需要阻塞最长使用率的进程就可以了
    receive.terminate()

Die laufenden Ergebnisse lauten wie folgt:

Aus der obigen Abbildung können wir sehen, dass sowohl send- als auch send_all-Prozesse Nachrichten über das von der Warteschlange instanziierte Warteschlangenobjekt senden können gleiche Empfangsfunktion Die von den beiden Prozessen übergebenen Nachrichten werden ebenfalls ausgedruckt.

Abschnitt

In diesem Kapitel haben wir Warteschlangen erfolgreich eingesetzt, um eine prozessübergreifende Kommunikation zu erreichen, und auch die Betriebsfähigkeiten von Warteschlangen gemeistert. In einer Warteschlange fügt ein Ende (hier demonstrieren wir das Sendeende) relevante Informationen über die Put-Methode hinzu, und das andere Ende verwendet die Get-Methode, um relevante Informationen abzurufen, um die Wirkung eines Prozesses zu erzielen Kommunikation.

Zusätzlich zu Warteschlangen können Prozesse auch über Pipes, Semaphoren und Shared Memory kommunizieren. Wenn Sie interessiert sind, können Sie sich über diese Methoden informieren. Sie können es selbst erweitern.

Andere Möglichkeiten der Kommunikation zwischen Prozessen – Ergänzung

Python bietet eine Vielzahl von Möglichkeiten der Prozesskommunikation, einschließlich Signale, Pipes, Nachrichtenwarteschlangen, Semaphoren, gemeinsam genutzter Speicher, Sockets usw.

Die beiden wichtigsten Methoden sind Warteschlange und Pipe . Warteschlange Wird zur Implementierung der Kommunikation zwischen mehreren Prozessen verwendet. Pipe ist die Kommunikation zwischen zwei Prozessen.

1. Pipes: unterteilt in anonyme Pipes und benannte Pipes

Anonyme Pipes: Beantragen Sie einen Puffer mit fester Größe im Kernel. Im Allgemeinen wird die Fock-Funktion verwendet, um die Kommunikation zwischen ihnen zu realisieren Übergeordnete und untergeordnete Prozesse

Benannte Pipe: Beantragen Sie einen Puffer mit fester Größe im Speicher. Das Programm hat das Recht, auch zwischen Prozessen zu kommunizieren, die nicht miteinander verbunden sind.

Eigenschaften: Orientiert an Byte-Streams. Der Lebenszyklus folgt dem Kernel und ist mit einem synchronen gegenseitigen Ausschlussmechanismus ausgestattet. Zwei Pipes realisieren eine bidirektionale Kommunikation. Eine Möglichkeit zum Umschreiben besteht darin, eine Warteschlange im Betriebssystemkern einzurichten, die die Warteschlange enthält Mehrere Datagrammelemente und mehrere Prozesse können über ein bestimmtes Handle auf die Warteschlange zugreifen. Nachrichtenwarteschlangen können verwendet werden, um Daten von einem Prozess an einen anderen zu senden. Es wird davon ausgegangen, dass jeder Datenblock einen Typ hat, und die vom Empfängerprozess empfangenen Datenblöcke können unterschiedliche Typen haben. Nachrichtenwarteschlangen weisen auch die gleichen Mängel wie Pipes auf, dh es gibt eine Obergrenze für die maximale Länge jeder Nachricht, eine Obergrenze für die Gesamtzahl der Bytes in jeder Nachrichtenwarteschlange und auch eine Obergrenze für die Gesamtzahl der Nachrichtenwarteschlangen im System

Merkmale: Nachrichtenwarteschlange Es kann als globale verknüpfte Liste betrachtet werden. Der Typ und der Inhalt des Datagramms werden in den Knoten der verknüpften Liste gespeichert, die mit der Kennung der Nachrichtenwarteschlange gekennzeichnet sind ; Die Nachrichtenwarteschlange ermöglicht es einem oder mehreren Prozessen, Nachrichten zu schreiben oder zu lesen. Der Zyklus folgt dem Kernel und kann eine bidirektionale Kommunikation erreichen

3. Semaphor: Erstellen Sie eine Semaphorsammlung (im Wesentlichen ein Array) im Kernel. Die Elemente des Arrays (Semaphore) sind alle 1. Verwenden Sie die P-Operation, um -1 auszuführen, und verwenden Sie die V-Operation, um +1

P auszuführen (sv): Wenn der Wert von sv größer als Null ist, dekrementieren Sie ihn um 1; wenn sein Wert Null ist, unterbrechen Sie die Ausführung des Programms Wenn kein Prozess aufgrund des Wartens auf sv hängen bleibt, fügen Sie 1 hinzu. Der PV-Vorgang wird für denselben Prozess verwendet, um einen gegenseitigen Ausschluss zu erreichen. Funktion: Schützen Sie kritische Ressourcen 4. Gemeinsam genutzter Speicher: Ordnen Sie denselben Teil des physischen Speichers dem virtuellen Adressraum verschiedener Prozesse zu, um die gemeinsame Nutzung derselben Ressource zwischen verschiedenen Prozessen zu erreichen. Wenn es um Kommunikationsmethoden zwischen Prozessen geht, kann Shared Memory als die nützlichste und schnellste Form von IPC bezeichnet werden.

Eigenschaften: Im Gegensatz zum häufigen Wechseln und Kopieren von Daten vom Benutzermodus in den Kernelmodus können Sie sie einfach direkt aus dem Speicher lesen ; gemeinsame Nutzung Der Speicher ist eine kritische Ressource, daher muss die Atomizität gewährleistet sein, wenn Vorgänge erforderlich sind. Sie können ein Semaphor oder einen Mutex verwenden.