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 分布式进程

相关文章