首页 > 代码库 > python 分布式进程体验

python 分布式进程体验

  抽了点时间体验了一把python 分布式进程,有点像分布式计算的意思,不过我现在还没有这个需求,先把简单体验的脚本发出来,供路过的各位高手指教

注:需要先下载multiprocessing 的python包支持才行。

管理端:

cat  task_manager.py  

#!/usr/bin/env python

#coding:utf8




import random,time,Queue,json

from multiprocessing.managers import BaseManager




#发送任务的队列:


task_queue = Queue.Queue()


#接收结果的队列:


result_queue = Queue.Queue()


#从BaseManager 继承的QueueManager:

class QueueManager(BaseManager):

pass



#把两个Queue 都注岫到网络上,callable 参数关联了Queue对象:

QueueManager.register(‘get_task_queue‘,callable=lambda:task_queue)

QueueManager.register(‘get_result_queue‘,callable=lambda:result_queue)


#绑定端口5000 设置验证码‘123456‘


manager = QueueManager(address=(‘‘,5000),authkey=‘123456‘)


#启动Queue:

manager.start()


#获得通过网络访问的Queue对象:

task = manager.get_task_queue()

result = manager.get_result_queue()


#放任务进去:

print ‘put tasks ........‘

shell_cmd = [‘date‘,‘hostname‘,‘uptime‘]

print ‘Put task \033[1;31;2m [%s] \033[0m..‘ %shell_cmd


while True:

print ‘get result.....‘

task.put(shell_cmd)

r = result.get()

print ‘Result:\033[1;32;40m %s \033[0m...‘ %r


manager.shutdown()


客户端:


#!/usr/bin/env python

#coding:utf8


import time,sys,Queue,os,commands

from multiprocessing.managers import BaseManager


#创建类似的QueueManager:

class QueueManager(BaseManager):

pass



#由于这个queuemanager 只从网络上获取queue,所以注册时只提供名字:

QueueManager.register(‘get_task_queue‘)

QueueManager.register(‘get_result_queue‘)


#连接到服务器 也就是迍务器

server_addr = ‘10.1.10.15‘

print ‘Connect to server %s...‘ %server_addr


#端口和验证码注意保持与taskmanager.py的一致

m = QueueManager(address=(server_addr,5000),authkey=‘123456‘)

#从网络连接

m.connect()

#获取Queue的对象

task = m.get_task_queue()

result = m.get_result_queue()


#从task队列取作务,并把结果写处result队列


n = task.get(timeout=1)

print n,type(n)


s,v =commands.getstatusoutput("ifconfig  | grep ‘inet addr:‘| grep -v ‘127.0.0.1‘ | cut -d: -f2 | awk ‘{ print $1}‘")

r_dic = {v:[]}

for i in n:


print ‘run task \033[1;36;40m %s \033[0m‘ %i

r = os.popen(i).read().split(‘\n‘)[:-1]

r_dic[v].append(r)

time.sleep(1)

result.put(r_dic)

print r_dic

#处理结束:

print ‘worker exit.‘