python 学习 -- 分布式进程

服务器端：

import random,time,queue

from multiprocessing.managers import BaseManager

# 发送任务的队列
task_queue = queue.Queue()
# 接收结果的队列
result_queue = queue.Queue()

class QueueManager(BaseManager):
pass

# 定义一个返回task_queue的函数
def return_task_queue():
global task_queue
return task_queue

# 定义一个返回结果队列的函数
def return_result_queue():
global result_queue
return result_queue

# 测试函数
def test():
# 将两个queue注册到网络上
QueueManager.register('get_task_queue', callable=return_task_queue)
QueueManager.register('get_result_queue', callable=return_result_queue)
# 绑定端口，同时设置验证码
manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
# 启动queue
manager.start()

# 获得通过网络访问的queue对象
task = manager.get_task_queue()
result = manager.get_result_queue()

# 放几个任务进去
for i in range(10):
n = random.randint(0, 100)
print('Put task %d...' % n)
task.put(n)

# 从result队列中取出结果
print("Try get results...")
for i in range(10):
r = result.get(timeout=10)
print("Result: %s" % r)

# 关闭(之前一直阻塞)
manager.shutdown()
print("master exit.")

# main函数
if __name__=="__main__":
test()

测试端：

import time,sys,queue
from multiprocessing.managers import BaseManager

class QueueManager(BaseManager):
pass

# 从网络上获取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,5000),authkey=b'abc')
m.connect()

# 获取queue对象
task = m.get_task_queue()
result = m.get_result_queue()
for i in range(10):
try:
n = task.get(timeout=1)
print('run task %d*%d...' % (n,n))
r = '%d * %d = %d' %(n,n,n*n)
time.sleep(1)
result.put(r)
except BaseException:
print('error')
print("worker exit.")