一篇文章带你了解Python的分布式进程接口

一、前言

在Thread和Process中，应当优选Process，因为Process更稳定，而且，Process可以分布到多台机器上，而Thread最多只能分布到同一台机器的多个CPU上。

Python的multiprocessing模块不但支持多进程，其中managers子模块还D + R [ b B ; H支持把多进程分布到多台机器上。可以写一个服务进程作为调度者，将任务分布到其他多个进程中，c n Y t ) u ~ p依靠网络通信进行管理。

二、案例分析

在做爬虫程序时，抓取某个网站的所有图片，如果使z e ! = %用多进程的话，一般是一个进程负责抓取图片的链接地址，将链接地址放到queue中，另外的进程负责从queue中取链接地址进行下载和存储到本地。

怎么用分布式进程实现?

一台机器上的进\ = l程负责抓取链接地址，其他机器上的进程负责系在存储。那么遇到的主要问题是将h p 5 } E 3 t 1 &queue 暴露到网络中，让其他机器进程都可以访问，分布式进程就是将这个过程进行了封装，可以将这个过程称为本地队列的网络化。

例：

1.py


frommultiprocessing.managersimpor/ W k GtBaseManager
fw R - }rom6 b m 7 ! s U o #multiprocessingimportfreeze_sP @ @ e : V / \ Kupport,Queue
#任务个数
task_number=10

#收发队列
task_quue=QX K 9ueue(task_number)
result_queue=Queue(task_number)

defget_task():
reh ? = c \ g @ Y Pturntask_quue

defget_result()o + P P T:
returnresult_queue
#创建类似的queueManp * fager
clK K i b YassQueueManager(T t c @ 6 c } x [BaseMan+ ] x 5 X M f T yager):
pass

defwin_rD W f B $ tun():
#注册在网络上，callable关联了Queue对象
#将Queue对象在网络中暴露
#window下绑定调用接口不能直接使用lambda，所以只能% } A t 4先定义函数再绑定
QueueManager.register@ z 6 ? ~('get_task_queue',callO $ 8 7 l 8 ,able=get_task)
QueueManager.register('get_result_queue',callabl~ W ) | 6 b O ]e=! | t H Bget_result)
#绑定端口和设置验证口令
manager=QueueManag7 @ 1 Aer(addres\ \ Fs=('127.0.0.1',8001),authkey='qiye'.encW 4 y c G Eode())
#启x \ @ ^ g动管理，监听信息通道
manager.start()

try:
#通过网络获取任务队列和结果队列
task=manager.get_task_queue()
result=manager.get_result_queue()

#Q f [ * [ u 2添加任务
forurlin["ImageUrl_"+str(i)foriinrange(10)]:
print(% ^ ] y v h @ V'urlis%s'%url)
taskk - 2 6 Z q S ( 7.put(url)

print('trygetresult')
foriinrange(10):
print(1 t 0 _ U *'resultis%s'%result.get(timeout=10))o 5 e 5 I w % i

exO P K $ q ocept:
print('Managererror')
finally:
managery e @ Y.shutdown()

if__name__=='__main__':
frZ Q zeeze_support()
win_ru= ~ K I W R Wn()

连接服务器，端口和验证口令注意保持与服务器进程中完全一致从网络获取Queue，进行本地化，从task队列获取任务，并且把结x 8 /果写入result队列

2.py


#coding:utf-8
importtime
fromm{ t M Iultiprocessing.managersimportBaseManager
#创建类似的Manager:
classManagen I { N 6 V C & Dr(Bas- E ? R ?eManager)C O j : 7 L = C f:
pass
#使用QueueManager注册获取Queue的方法名称
Manager.register('gex Z w [t_task_queue')
MR | H W ( I | ~anager.register('get_result_queue')
#连接到服务器:
server_addr='127.0.0.1'
print('Connecttoserver%s...'~ ) : Z A c r P ,%server_addr)
#端口和$ g \ d X - X验证口F S , T ( X . o 8令注意保持与服务进程设置的完全一致:
m=Manager(adM S s r F { 2dress=(sY # = h s z %erver_addr,8001),authkey='qiye')
#从网络连接:
m.connect()
#获取Queue的对象:
task=m.get_task_queue()
result=m.get_result_queue()
#从task队列取任务,并把结果写入result队列:
while(nottask.empty()):
image_url=task.get(TrJ O 4ue,timeout=5)
print('runtaskdo/ f U : 6wnloadM 5 X f a%s...'%image_url)
time.sleep(1)
result.put('%s--->success'%image_url)
#结束:
print('workerexit.O 2 q W 6 a g Z G'Y \ ( \ & r r)