Qu'est-ce qu'un processus distribué Python ? (Exemple d'analyse)

Dans cet article, examinons ce qu'est le processus distribué python. Découvrez les processus distribués Python et le rôle que les processus distribués peuvent jouer dans la programmation Python.

Parmi Thread et Process, Process doit être préféré car Process est plus stable et Process peut être distribué sur plusieurs machines, tandis que Thread ne peut être distribué qu'à plusieurs CPU sur la même machine au maximum.

Le module multitraitement de Python prend non seulement en charge plusieurs processus, mais le sous-module managers prend également en charge la distribution de plusieurs processus sur plusieurs machines. Un processus de service peut agir comme un planificateur, distribuant des tâches à plusieurs autres processus, en s'appuyant sur la communication réseau. Le module gestionnaires étant bien encapsulé, vous pouvez facilement écrire des programmes multi-processus distribués sans connaître les détails de la communication réseau.

Par exemple : si nous avons déjà un programme multi-processus qui communique via Queue s'exécutant sur la même machine, maintenant, parce que le processus qui gère la tâche a une lourde charge de travail, nous voulons séparer le processus qui envoie la tâche et le processus qui gère la tâche. Distribué sur deux machines. Comment l'implémenter à l'aide d'un processus distribué ?

La file d'attente d'origine peut continuer à être utilisée, mais en exposant la file d'attente via le réseau via le module gestionnaires, les processus sur d'autres machines peuvent accéder à la file d'attente.

Examinons d'abord le processus de service. Le processus de service est responsable du démarrage de la file d'attente, de l'enregistrement de la file d'attente sur le réseau, puis de l'écriture des tâches dans la file d'attente :

# task_master.py

import random, time, queue
from multiprocessing.managers import BaseManager
# 发送任务的队列:
task_queue = queue.Queue()
# 接收结果的队列:
result_queue = queue.Queue()
# 从BaseManager继承的QueueManager:
class QueueManager(BaseManager):
    pass
# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 启动Queue:
manager.start()
# 获得通过网络访问的Queue对象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放几个任务进去:
for i in range(10):
    n = random.randint(0, 10000)
    print('Put task %d...' % n)
    task.put(n)
# 从result队列读取结果:
print('Try get results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result: %s' % r)
# 关闭:
manager.shutdown()

Veuillez noter que lorsque nous écrivons un programme multi-processus sur la machine, la file d'attente créée peut être utilisée directement. Cependant, dans un environnement multi-processus distribué, l'ajout de tâches à la file d'attente ne peut pas fonctionner directement sur la file d'attente de tâches d'origine, qui contourne. l'encapsulation du QueueManager. Il doit être ajouté via l'interface Queue obtenue par manager.get_task_queue().

Ensuite, démarrez le processus de tâche sur une autre machine (il peut également être démarré sur cette machine) :

# task_master.py
import random, time, queue
from multiprocessing.managers import BaseManager
# 发送任务的队列:
task_queue = queue.Queue()
# 接收结果的队列:
result_queue = queue.Queue()
# 从BaseManager继承的QueueManager:
class QueueManager(BaseManager):
    pass
# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 启动Queue:
manager.start()
# 获得通过网络访问的Queue对象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放几个任务进去:
for i in range(10):
    n = random.randint(0, 10000)
    print('Put task %d...' % n)
    task.put(n)
# 从result队列读取结果:
print('Try get results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result: %s' % r)
# 关闭:
manager.shutdown()

Le processus de tâche doit se connecter au processus de service via le réseau, donc le processus de service doit être spécifié IP.

Maintenant, vous pouvez essayer l'effet fonctionnel du processus distribué. Démarrez d'abord le processus de service task_master.py :

$ python3 task_master.py 
Put task 3411...
Put task 1605...
Put task 1398...
Put task 4729...
Put task 5300...
Put task 7471...
Put task 68...
Put task 4219...
Put task 339...
Put task 7866...
Try get results...

Une fois que le processus task_master.py a envoyé la tâche, il commence à attendre les résultats de la file d'attente des résultats. Démarrez maintenant le processus task_worker.py :

$ python3 task_worker.pyConnect to server 127.0.0.1...
run task 3411 * 3411...
run task 1605 * 1605...
run task 1398 * 1398...
run task 4729 * 4729...
run task 5300 * 5300...
run task 7471 * 7471...
run task 68 * 68...
run task 4219 * 4219...
run task 339 * 339...
run task 7866 * 7866...
worker exit.

le processus task_worker.py se termine et les résultats continueront à être imprimés dans le processus task_master.py :

Result: 3411 * 3411 = 11634921
Result: 1605 * 1605 = 2576025
Result: 1398 * 1398 = 1954404
Result: 4729 * 4729 = 22363441
Result: 5300 * 5300 = 28090000
Result: 7471 * 7471 = 55815841
Result: 68 * 68 = 4624
Result: 4219 * 4219 = 17799961
Result: 339 * 339 = 114921
Result: 7866 * 7866 = 61873956

Ce simple Master /Modèle Worker A quoi ça sert ? En fait, il s'agit d'un calcul simple mais véritablement distribué. En modifiant légèrement le code et en démarrant plusieurs travailleurs, les tâches peuvent être réparties sur plusieurs voire plusieurs dizaines de machines. Par exemple, le code de calcul de n*n peut être remplacé par l'envoi. emails. , réalisant l'envoi asynchrone des files d'attente de courrier.

La raison pour laquelle la file d'attente est accessible via le réseau est via QueueManager. Puisque QueueManager gère plus d'une file d'attente, il est nécessaire de donner un nom à l'interface d'appel réseau de chaque file d'attente, tel que get_task_queue.

À quoi sert la clé d'authentification ? Cela permet de garantir que les deux machines communiquent normalement et ne soient pas perturbées de manière malveillante par d'autres machines. Si la clé d'authentification de task_worker.py est incohérente avec la clé d'authentification de task_master.py, la connexion échouera définitivement.

L'interface de processus distribué de Python est simple et bien encapsulée, ce qui la rend adaptée aux environnements dans lesquels des tâches lourdes doivent être distribuées sur plusieurs machines.

Notez que le rôle de Queue est de livrer des tâches et de recevoir des résultats. La quantité de données décrivant chaque tâche doit être aussi petite que possible. Par exemple, lors de l'envoi d'une tâche pour traiter un fichier journal, au lieu d'envoyer le fichier journal lui-même, qui fait plusieurs centaines de mégaoctets, envoyez le chemin complet où le fichier journal est stocké, et le processus Worker lira ensuite le fichier depuis le disque partagé.

Ce qui précède est tout le contenu décrit dans cet article. Cet article présente principalement les connaissances pertinentes du processus distribué python. J'espère que vous pourrez utiliser les informations pour comprendre le contenu ci-dessus. J'espère que ce que j'ai décrit dans cet article vous sera utile et vous facilitera l'apprentissage de Python.

Pour plus de connaissances connexes, veuillez visiter la colonne Tutoriel Python du site Web php chinois.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!