Detailed introduction to distributed processes in python (with examples)

The content of this article is about what is SAPI in PHP? How to achieve? (Pictures and text), it has certain reference value. Friends in need can refer to it. I hope it will be helpful to you.

Among Thread and Process, Process should be preferred because Process is more stable and Process can be distributed to multiple machines, while Thread can only be distributed to multiple CPUs on the same machine at most. .

Python's multiprocessing module not only supports multiple processes, but the managers submodule also supports distributing multiple processes to multiple machines. A service process can act as a scheduler, distributing tasks to multiple other processes, relying on network communication. Since the managers module is well encapsulated, you can easily write distributed multi-process programs without knowing the details of network communication.

Exposing the Queue through the network through the managers module allows processes on other machines to access the Queue. Let’s look at the service process first. The service process is responsible for starting the Queue, registering the Queue on the network, and then writing tasks into the Queue.

BaseManager: Provides a way to share data between different machine processes;

(重要的点: 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()

When you run the program, you will wait for the execution result for 10s. If there is no worker to obtain the task, the result will be returned and the program will Report an error.

When we write a multi-process program on one machine, the created Queue can be used directly. However, in distributed multi-process In this environment, adding tasks to the Queue cannot directly operate on the original task_queue, which bypasses the encapsulation of the QueueManager. It must be added through the Queue interface obtained by 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.')

Use the command line to run the program, the results are more intuitive

The above is the detailed content of Detailed introduction to distributed processes in python (with examples). For more information, please follow other related articles on the PHP Chinese website!