Python의 분산 프로세스에 대한 자세한 소개(예제 포함)

이 글의 내용은 PHP에서 SAPI가 무엇인지에 관한 것입니다. 달성하는 방법? (사진 및 텍스트) 특정 참고 가치가 있습니다. 도움이 필요한 친구들이 참고할 수 있기를 바랍니다.

Thread와 Process 중 Process를 선호하는 이유는 Process가 더 안정적이고 Process를 여러 머신에 배포할 수 있는 반면 Thread는 최대 동일한 머신의 여러 머신에만 배포할 수 있기 때문입니다. .

Python의 다중 처리 모듈은 다중 프로세스를 지원할 뿐만 아니라 관리자 하위 모듈도 다중 프로세스를 다중 컴퓨터에 배포하는 것을 지원합니다. 서비스 프로세스는 네트워크 통신에 의존하여 작업을 여러 다른 프로세스에 배포하는 스케줄러 역할을 할 수 있습니다. 관리자 모듈은 잘 캡슐화되어 있으므로 네트워크 통신의 세부 사항을 몰라도 분산 다중 프로세스 프로그램을 쉽게 작성할 수 있습니다.

관리자 모듈을 통해 네트워크를 통해 대기열을 노출하면 다른 컴퓨터의 프로세스가 대기열에 액세스할 수 있습니다. 먼저 서비스 프로세스를 살펴보겠습니다. 서비스 프로세스는 대기열을 시작하고 네트워크에 대기열을 등록한 다음 대기열에 작업을 쓰는 역할을 합니다.

BaseManager: 서로 다른 머신 프로세스 간에 데이터를 공유하는 방법을 제공합니다.

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

프로그램을 실행하면 10초 동안 실행 결과를 기다립니다. 작업자가 작업을 가져오려면 반환해야 합니다. 결과적으로 프로그램은 오류를 보고합니다.

머신에서 다중 프로세스 프로그램을 작성할 때 생성된 Queue를 직접 사용할 수 있습니다. 분산 다중 프로세스 환경에서 대기열에 작업을 추가하는 것은 원래 task_queue에서 직접 작동할 수 없습니다. 이는 QueueManager의 캡슐화를 우회하고 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.')

명령줄을 사용하여 프로그램을 실행하면 결과가 더욱 직관적입니다.

#🎜🎜 #

위 내용은 Python의 분산 프로세스에 대한 자세한 소개(예제 포함)의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!