Introduction détaillée aux processus distribués en python (avec exemples)

Le contenu de cet article porte sur qu'est-ce que SAPI en PHP ? Comment y parvenir ? (Images et texte), il a une certaine valeur de référence. Les amis dans le besoin peuvent s'y référer. J'espère qu'il vous sera utile.

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 de la même machine au maximum.

Le module multitraitement de Python prend non seulement en charge plusieurs processus, mais le sous-module gestionnaires 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.

L'exposition de la file d'attente via le réseau via le module des gestionnaires permet aux processus sur d'autres machines d'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.

BaseManager : Fournit un moyen de partager des données entre différents processus machine

(重要的点: ip:port)

# task_master.py

import random
from multiprocessing import freeze_support
from queue import Queue
from multiprocessing.managers import  BaseManager
# 1. 创建需要的队列
# task_queue：发送任务的队列
# coding=utf-8

import random,time
from queue import Queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support

task_queue =  Queue()  # 发送任务的队列:
result_queue = Queue() # 接收结果的队列:
class QueueManager(BaseManager):  # 从BaseManager继承的QueueManager:
    pass
# windows下运行
def return_task_queue():
    global task_queue
    return task_queue  # 返回发送任务队列
def return_result_queue ():
    global result_queue
    return result_queue # 返回接收结果队列

def test():
    # 把两个Queue都注册到网络上, callable参数关联了Queue对象,它们用来进行进程间通信，交换对象
    #QueueManager.register('get_task_queue', callable=lambda: task_queue)
    #QueueManager.register('get_result_queue', callable=lambda: result_queue)
    QueueManager.register('get_task_queue', callable=return_task_queue)
    QueueManager.register('get_result_queue', callable=return_result_queue)
    # 绑定端口4000, 设置验证码'sheenstar':
    #manager = QueueManager(address=('', 4000), authkey=b'sheenstar')
    # windows需要写ip地址
    manager = QueueManager(address=('192.168.1.160', 4000), authkey=b'sheenstar')
    manager.start()  # 启动Queue:
    # 获得通过网络访问的Queue对象:
    task = manager.get_task_queue()
    result = manager.get_result_queue()
    for i in range(13):   # 放几个任务进去:
        n = random.randint(0, 10000)
        print('Put task %d...' % n)
        task.put(n)
    # 从result队列读取结果:
    print('Try get results...')
    for i in range(13):
        r = result.get(timeout=10)
        print('Result: %s' % r)

    # 关闭:
    manager.shutdown()
    print('master exit.')
if __name__=='__main__':
    freeze_support()
    print('start!')
    test()

Lorsque vous exécutez le programme, vous attendez le résultat de l'exécution pendant 10 secondes ; il n'y a aucun travailleur pour obtenir la tâche, renvoyez le résultat et le programme signalera une erreur.

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

# coding=utf-8
import time, sys
from queue import Queue
from multiprocessing.managers import BaseManager

# 创建类似的QueueManager:
class QueueManager(BaseManager):
    pass

# 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

# 连接到服务器，也就是运行task_master.py的机器:
server_addr = '192.168.1.160'
print('Connect to server %s...' % server_addr)
# 端口和验证码注意保持与task_master.py设置的完全一致:
m = QueueManager(address=(server_addr, 4000), authkey=b'sheenstar')
# 从网络连接:
try:
    m.connect()
except:
    print('请先启动task_master.py!')
    #sys.exit("sorry, goodbye!");
# 获取Queue的对象:
task = m.get_task_queue()
result = m.get_result_queue()
# 从task队列取任务,并把结果写入result队列:
for i in range(13):
    try:
        n = task.get()
        print('run task %d * %d...' % (n, n))
        r = '%d * %d = %d' % (n, n, n*n)
        time.sleep(1)
        result.put(r)
    except ConnectionResetError as e:
        print("任务执行结束，自动断开连接")
# 处理结束:
print('worker exit.')

Utilisez la ligne de commande pour exécuter le programme, les résultats sont plus intuitifs

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!