二十五、创建分布式进程

1、概念

  分布式进程指的是将Process进程分布到多台机器上,充分利用多台机器的性能完成复杂的任务。

  multiprocessing模块不仅支持多进程,而且其中的managers模块还支持把多进程分布到多台机器上。例如,写一个服务进程作为调度者,将任务分布到其他多个进程中,依靠网络通信(将Queue暴露到网络中,实现本地队列的网络化)进行管理。

2、步骤

  2.1  服务进程的创建

  1、建立队列Queue,用来进行进程间的通信。

    (1)服务进程创建任务队列task_queue,用来作为传递任务给任务进程的通道

    (2)服务进程创建结果队列result_queue,用来作为任务进程完成任务后回复服务进程的通道

  2、把创建的队列在网络上注册,暴露给其他进程

    (1)通过multiprocessing.managers.BaseManage.register创建Queue接口来作为添加任务的通道,typeid是调用的方法名,callable是绑定的本地获取Queue的方法

  3、创建一个对象(multiprocessing.managers.BaseManage)的实例manager,绑定端口和验证口令

    (1)通过multiprocessing.managers.BaseManage绑定本地端口,指定authkey,authkey接受bytes类型

  4、启动manager,开始监听信息通道

    (1)manager.start()

  5、通过管理实例的方法(第2步中注册的typeid的参数)获得通过网络访问的Queue对象,即再把网络队列实例化成可以使用的本地队列

    (1)task = manager.get_task_queue()

  6、创建任务到本地队列中,会自动上传到网络队列中,分配给任务进程去处理

    (1)task.put()

  7、关闭实例manager

    (1)manager.shutdown()

  8、示例代码

# 示例代码来自《Python爬虫开发与项目实战》,作者范传辉

from multiprocessing.managers import BaseManager
import queue


# 创建队列
task_queue = queue.Queue()
result_queue = queue.Queue()

# 注册方法,在网络上暴露队列
# QueueManager.register(typeid="get_task_queue", callable=lambda: task_queue)
# QueueManager.register(typeid=‘get_result_queue‘, callable=lambda: result_queue)
BaseManager.register(typeid="get_task_queue", callable=lambda: task_queue)
BaseManager.register(typeid=get_result_queue, callable=lambda: result_queue)

# 创建本例manager实例
address = (127.0.0.1, 8001)
# manager = QueueManager(address=address, authkey=b‘biu‘)
manager = BaseManager(address=address, authkey=bbiu)

# 启动
manager.start()

# 通过管理实例的方法获得通过网络访问的Queue对象
task = manager.get_task_queue()
result = manager.get_result_queue()

# 添加任务
for url in ["image_url_" + str(i) for i in range(10)]:
    print(put task %s... % url)
    task.put(url)

# 获取返回结果
print("try get result...")
for i in range(10):
    print(result in %s  % result.get(timeout=10))

manager.shutdown()

  2.2 任务进程的创建

  1、使用multiprocessing.managers.BaseManager注册用于获取Queue的方法名称(任务进程只能通过名称来在网络上获取Queue)

    (1)BaseManger.register(typeid=‘get_task_queue‘)

  2、连接服务器,端口和验证口令要与服务进程完全一致

    (1)创建manager实例:manager = BaseManager(address=address,authkey=b‘...‘)

    (2)连接服务器:manager.connect()

  3、本地化网络队列

    (1)task = manager.get_task_queue()

  4、从任务队列获取任务,将结果写入result队列

  5、实例代码

# 示例代码来自《Python爬虫开发与项目实战》,作者范传辉
import time
from multiprocessing.managers import BaseManager

# 注册用于获取queue的方法名称
BaseManager.register(typeid=get_task_queue)
BaseManager.register(typeid=get_result_queue)

# 连接到服务器
server_addr = (127.0.0.1, 8001)
print("connect to server %s" % str(server_addr))
manager = BaseManager(address=server_addr, authkey=bbiu)
manager.connect()

# 获取queue对象
task = manager.get_task_queue()
result = manager.get_result_queue()

while not task.empty():
    image_url = task.get(True,timeout=5)
    print(run task download %s... % image_url)
    time.sleep(1)
    result.put("%s ---> success" % image_url)

# 处理结束
print(worker exit)

 

    

    

 

二十五、创建分布式进程

上一篇:Mac安装cnpm


下一篇:1085 Perfect Sequence (25 分)【简单 / 双指针】