python分布式进程

1、master端代码

 1 import random, time, queue
 2 from multiprocessing.managers import BaseManager
 3 
 4 # 发送任务的队列
 5 task_queue = queue.Queue()
 6 # 接受任务的队列
 7 result_queue = queue.Queue()
 8 
 9 
10 class QueueManager(BaseManager):
11     pass
12 
13 
14 # 把两个Queue都注册到网络上，allable参数关联了Queue对象:
15 QueueManager.register('get_task_queue', callable=lambda: task_queue)
16 QueueManager.register('get_result_queue', callable=lambda: result_queue)
17 
18 # 绑定端口5000, 设置验证码'abc':
19 manager = QueueManager(address=('', 5000), authkey=b'abc')
20 # 启动Queue
21 manager.start()
22 # 获得通过网络访问的Queue对象：
23 task = manager.get_task_queue()
24 result = manager.get_result_queue()
25 
26 # 放几个任务进去：
27 for i in range(1000):
28     n = random.randint(0, 1000)
29     print('Put task %d ...' % n)
30     task.put(n)
31 
32 # 从result队列中读取结果：
33 print('Try get results...')
34 for i in range(1000):
35     r = result.get(timeout=10)
36     print('Result :%s' % r)
37 
38 #关闭
39 manager.shutdown()
40 print('master exit.')

2、slave代码

 1 # task_worker.py
 2 
 3 import time, sys, queue
 4 from multiprocessing.managers import BaseManager
 5 
 6 # 创建类似的QueueManager:
 7 class QueueManager(BaseManager):
 8     pass
 9 
10 # 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字:
11 QueueManager.register('get_task_queue')
12 QueueManager.register('get_result_queue')
13 
14 # 连接到服务器，也就是运行task_master.py的机器:
15 server_addr = '127.0.0.1'
16 print('Connect to server %s...' % server_addr)
17 # 端口和验证码注意保持与task_master.py设置的完全一致:
18 m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
19 # 从网络连接:
20 m.connect()
21 # 获取Queue的对象:
22 task = m.get_task_queue()
23 result = m.get_result_queue()
24 # 从task队列取任务,并把结果写入result队列:
25 for i in range(100):
26     try:
27         n = task.get(timeout=1)
28         print('run task %d * %d...' % (n, n))
29         r = '%d * %d = %d' % (n, n, n*n)
30         time.sleep(1)
31         result.put(r)
32     except task.Empty:
33         print('task queue is empty.')
34 # 处理结束:
35 print('worker exit.')

3、先执行master，再执行slave