分布式进程 - 爱码网

#分布式进程

分布式进程

#分布式进程六个步骤：
#1.建立队列Queue，用来进程间的通信。
#2.将队列Queue进行网络共享给其他进程
#3.建立对象实例，绑定端口和验证口令
#4.启动对象实例进行信息传输管理
#5.通过对象实例将网络队列本地化
#6.分配本地化队列进行进程任务处理

# 调度服务进程taskmanager.py
import random,time,Queue
from multiprocessing.managers import BaseManager
#第一步，建立task_queue和result_queue，用来存放任务和结果
task_queue = Queue.Queue()
result_queue = Queue.Queue()

class Queuemanager(BaseManager):
    pass
#第二步，把创建的两个队列注册到网络，利用register方法，callable参数关联了Queue对象
#将Queue对象在网络上暴露
Queuemanager.register('get_task_queue',callable=lambda:task_queue)
Queuemanager.register('get_result_queue',callable=lambda:result_queue)
#第三步，绑定端口8001，设置验证口令‘qiye’相当于对象的初始化
manager = Queuemanager(address=('',8001),authkey='qiye')
#第四步，启动管理，监听信息通道
manager.start()
#第五步，通过管理实例的方法获得通过网络访问的Queue对象
task = manager.get_task_queue()
result = manager.get_result_queue()
#第六步，添加任务
for url in ['ImagerUrl'+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()

#任务进程taskWorker.py
import time
from multiprocessing.managers import BaseManager

#创建类似的QueueManager
calss QueueManager(BaseManager):
    pass
#第一步，使用QueueManager注册用于获取Queue的放法名称
QueueManager.register('get_task_queue')
QueueManager.resgister('get_result_queue')
#第二步，连接到服务器
server_addr = '127.0.0.1'
print('Connect to server %s...'%server_addr)
#端口和验证口令注意和服务进程保持一致
m = QueueManager(address=(server_addr,8001),authkey='qiye')
#链接网络
m.conncet()
#第三步，获取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---->>>success'%image_url)
#关闭处理
print('worker exit.')

分布式进程