Python 分布式进程

#-*-coding:utf-8-*-
'''分布式进程指的是将Process进程分不到多台机器上，充分利用多台机器的性能完成复杂的任务'''
#服务器端
#---------------------------------------Linux版----------------------------------------------
# import random,time,Queue
# from multiprocessing.managers import BaseManager
# #一、建立task_queue和result_queue用来存放任务和结果
# task_queue=Queue.Queue()
# result_queue=Queue.Queue()
# class Queuemanger(BaseManager):
#     pass
# #二、把创建的两个队列注册在网络上，利用reister方法，clallble参数关联了Queue对象，将Queue对象在网络中暴露
# Queuemanger.register('get_task_queue',callable=lambda:task_queue)
# Queuemanger.register('get_result_queue',callable=lambda:result_queue)
# #三、绑定端口8001，设置端口口令；admin，相当于对象的初始化
# manager=Queuemanger(address=('',8001),authkey='admin')
# #四、启动管理监听信息通道
# manager.start()
# #五、通过管理实例的方法获得通过网络访问的Queue对象
# task=manager.get_task_queue()
# result=manager.get_result_queue()
# #六、添加任务
# for url in ["ImageUrl_"+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 is %s"%result.get(timeout=10))
# #关闭管理
# manager.shutdown()
#--------------------------------------------------Windows版-----------------------------------
#taskManager for Windows
import Queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support
#任务个数
task_number=20
#定义收发队列
task_queue=Queue.Queue(task_number)
result_queue=Queue.Queue(task_number)
def get_task():
    return task_queue
def get_result():
    return task_queue()
#创建类似的QueueManager
class QueueManager(BaseManager):
    pass
def win_run():
    #windows下绑定调用接口不能使用lambda所以只能先定义函数再绑定
    QueueManager.register('get_task_queue',callable=get_task)
    QueueManager.register('get_result_queue',callable=get_result)
    #绑定端口并设置验证口令，Windows下需要填写IP地址，linux下不填写默认使用本地IP地址
    manager=QueueManager(address=('127.0.0.1',8001),authkey='admin')
    #启动
    manager.start()
    try:
        #通过网络获取任务队列和结果队列
        task=manager.get_task_queue()
        result=manager.get_result_queue()
        #添加任务
        for url in["ImageUrl_"+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 is %s '%result.get(timeout=10))
    except Exception as e:
        print('Manager error:%s'%e)
    finally:
        #不论程序执行成功或是失败finally都会执行，即一定要将管道关闭，否则汇报错误
        manager.shutdown()
if __name__=="__name__":
    #windows下多进程可能会有问题，添加以下代码可以缓解
    freeze_support()
    win_run()

#客户端

#-*-coding:utf-8-*-

#任务进程TaskWorker.py
import time
from multiprocessing.managers import BaseManager
#创建类似的QueueManager：
class QueueManager(BaseManager):
    pass
#一、使用QueueManger注册用于获取Queue的方法名称
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')
#二、链接到服务器
server_addr='127.0.0.1'
print('Connect to server %s...'%server_addr)
#端口和验证口令需要与服务器保持一致
m=QueueManager(address=(server_addr,8001),authkey='admin')
#从网络链接
m.connect()
#三、获取Queue的对象
task=m.get_task_queue()
result=m.get_result_queue()
#四、从 task队列获取任务，并把结果写入result
while(not task.empty()):
    image_url=task.get(True,timeout=5)
    print('run task download %s...'%image_url)
    time.sleep(1)
    result.put('%s--->sucess'%image_url)
print('worker exit.')

秒客网

Python 分布式进程

相关文章