Python分布式进程中你会遇到的问题解析

站长资源 2024/11/26 佚名

2 0 1

小惊大怪

你是不是在用Python3或者在windows系统上编程"color: #ff0000">分布式进程

正如大家所知道的Process比Thread更稳定，而且Process可以分布到多台机器上，而Thread最多只能分布到同一台机器的多个CPU上。Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程分布到多台机器上。一个服务进程可以作为调度者，将任务分布到其他多个进程中，依靠网络通信。由于managers模块封装很好，不必了解网络通信的细节，就可以很容易地编写分布式多进程程序。

代码记录

举个例子

如果我们已经有一个通过Queue通信的多进程程序在同一台机器上运行，现在，由于处理任务的进程任务繁重，希望把发送任务的进程和处理任务的进程分布到两台机器上，这应该怎么用分布式进程来实现呢"htmlcode">

#！/user/bin/pytthon 
# -*- coding:utf-8 -*- 
# @Time: 2018/3/3 16:46 
# @Author: lichexo 
# @File: task_master.py 
import random, time, queue 
from multiprocessing.managers import BaseManager 
# 发送任务的队列: 
task_queue = queue.Queue() 
# 接收结果的队列: 
result_queue = queue.Queue() 
# 从BaseManager继承的QueueManager: 
class QueueManager(BaseManager): 
 pass 
# 把两个Queue都注册到网络上, callable参数关联了Queue对象: 
QueueManager.register('get_task_queue', callable=lambda: task_queue) 
QueueManager.register('get_result_queue', callable=lambda: result_queue) 
# 绑定端口5000, 设置验证码'abc': 
manager = QueueManager(address=('', 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, 10000) 
 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.')

请注意，当我们在一台机器上写多进程程序时，创建的Queue可以直接拿来用，但是，在分布式多进程环境下，添加任务到Queue不可以直接对原始的task_queue进行操作，那样就绕过了QueueManager的封装，必须通过manager.get_task_queue()获得的Queue接口添加。然后，在另一台机器上启动任务进程(本机上启动也可以)

写个task_worker.py

#！/user/bin/pytthon 
# -*- coding:utf-8 -*- 
# @Time: 2018/3/3 16:46 
# @Author: lichexo 
# @File: task_worker.py 
import time, sys, queue 
from multiprocessing.managers import BaseManager 
# 创建类似的QueueManager: 
class QueueManager(BaseManager): 
 pass 
# 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字: 
QueueManager.register('get_task_queue') 
QueueManager.register('get_result_queue') 
# 连接到服务器，也就是运行task_master.py的机器: 
server_addr = '127.0.0.1' 
print('Connect to server %s...' % server_addr) 
# 端口和验证码注意保持与task_master.py设置的完全一致: 
m = QueueManager(address=(server_addr, 5000), authkey=b'abc') 
# 从网络连接: 
m.connect() 
# 获取Queue的对象: 
task = m.get_task_queue() 
result = m.get_result_queue() 
# 从task队列取任务,并把结果写入result队列: 
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 Queue.Empty: 
 print('task queue is empty.') 
# 处理结束: 
print('worker exit.')

任务进程要通过网络连接到服务进程，所以要指定服务进程的IP。

运行结果

现在，可以试试分布式进程的工作效果了。先启动task_master.py服务进程：

Traceback (most recent call last):
File "F:/Python/untitled/xianchengjincheng/master.py", line 25, in <module>
manager.start()
File "F:Pythonpystalllibmultiprocessingmanagers.py", line 513, in start
self._process.start()
File "F:Pythonpystalllibmultiprocessingprocess.py", line 105, in start
self._popen = self._Popen(self)
File "F:Pythonpystalllibmultiprocessingcontext.py", line 322, in _Popen
return Popen(process_obj)
File "F:Pythonpystalllibmultiprocessingpopen_spawn_win32.py", line 65, in __init__
reduction.dump(process_obj, to_child)
File "F:Pythonpystalllibmultiprocessing
eduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <function <lambda> at 0x00000202D1921E18>: attribute lookup <lambda> on __main__ failed

task_master.py进程发送完任务后，开始等待result队列的结果。现在启动task_worker.py进程：

Connect to server 127.0.0.1...
Traceback (most recent call last):
File "F:/Python/untitled/xianchengjincheng/work.py", line 24, in <module>
m.connect()
File "F:Pythonpystalllibmultiprocessingmanagers.py", line 489, in connect
conn = Client(self._address, authkey=self._authkey)
File "F:Pythonpystalllibmultiprocessingconnection.py", line 487, in Client
c = SocketClient(address)
File "F:Pythonpystalllibmultiprocessingconnection.py", line 614, in SocketClient
s.connect(address)
ConnectionRefusedError: [WinError 10061] 由于目标计算机积极拒绝，无法连接。

看到没，结果都出错了，我们好好分析一下到底哪出错了。。。

错误分析

在task_master.py的报错提示中，我们知道它说lambda错误，这是因为序列化不支持匿名函数，所以我们得修改代码，重新对queue用QueueManager进行封装放到网络中。

# 把两个Queue都注册到网络上, callable参数关联了Queue对象 
QueueManager.register('get_task_queue',callable=return_task_queue) 
QueueManager.register('get_result_queue',callable=return_result_queue)

其中task_queue和result_queue是两个队列，分别存放任务和结果。它们用来进行进程间通信，交换对象。

因为是分布式的环境，放入queue中的数据需要等待Workers机器运算处理后再进行读取，这样就需要对queue用QueueManager进行封装放到网络中，这是通过上面的2行代码来实现的。我们给return_task_queue的网络调用接口取了一个名get_task_queue,而return_result_queue的名字是get_result_queue，方便区分对哪个queue进行操作。task.put(n)即是对task_queue进行写入数据，相当于分配任务。而result.get()即是等待workers机器处理后返回的结果。

值得注意在windows系统中你必须要写IP地址，而其他操作系统比如linux操作系统则就不要了。

# windows需要写ip地址 
manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')

修改后的代码

在task_master.py中修改如下：

#！/user/bin/pytthon 
# -*- coding:utf-8 -*- 
# @Time: 2018/3/3 16:46 
# @Author: lichexo 
# @File: task_master.py 
# task_master.py 
import random,time,queue 
from multiprocessing.managers import BaseManager 
from multiprocessing import freeze_support 
task_queue = queue.Queue() # 发送任务的队列: 
result_queue = queue.Queue() # 接收结果的队列: 
class QueueManager(BaseManager): # 从BaseManager继承的QueueManager: 
 pass 
# windows下运行 
def return_task_queue(): 
 global task_queue 
 return task_queue # 返回发送任务队列 
def return_result_queue (): 
 global result_queue 
 return result_queue # 返回接收结果队列 
def test(): 
 # 把两个Queue都注册到网络上, callable参数关联了Queue对象,它们用来进行进程间通信，交换对象 
 #QueueManager.register('get_task_queue', callable=lambda: task_queue) 
 #QueueManager.register('get_result_queue', callable=lambda: result_queue) 
 QueueManager.register('get_task_queue', callable=return_task_queue) 
 QueueManager.register('get_result_queue', callable=return_result_queue) 
 # 绑定端口5000, 设置验证码'abc': 
 #manager = QueueManager(address=('', 5000), authkey=b'abc') 
 # windows需要写ip地址 
 manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc') 
 manager.start() # 启动Queue: 
 # 获得通过网络访问的Queue对象: 
 task = manager.get_task_queue() 
 result = manager.get_result_queue() 
 for i in range(10): # 放几个任务进去: 
 n = random.randint(0, 10000) 
 print('Put task %d...' % n) 
 task.put(n) 
 # 从result队列读取结果: 
 print('Try get results...') 
 for i in range(10): 
 # 这里加了异常捕获 
 try: 
 r = result.get(timeout=5) 
 print('Result: %s' % r) 
 except queue.Empty: 
 print('result queue is empty.') 
 # 关闭: 
 manager.shutdown() 
 print('master exit.') 
if __name__=='__main__': 
 freeze_support() 
 print('start!') 
 test()

在task_worker.py中修改如下：

#！/user/bin/pytthon 
# -*- coding:utf-8 -*- 
# @Time: 2018/3/3 16:46 
# @Author: lichexo 
# @File: task_worker.py 
# task_worker.py 
import time, sys, queue 
from multiprocessing.managers import BaseManager 
# 创建类似的QueueManager: 
class QueueManager(BaseManager): 
 pass 
# 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字: 
QueueManager.register('get_task_queue') 
QueueManager.register('get_result_queue') 
# 连接到服务器，也就是运行task_master.py的机器: 
server_addr = '127.0.0.1' 
print('Connect to server %s...' % server_addr) 
# 端口和验证码注意保持与task_master.py设置的完全一致: 
m = QueueManager(address=(server_addr, 5000), authkey=b'abc') 
# 从网络连接: 
m.connect() 
# 获取Queue的对象: 
task = m.get_task_queue() 
result = m.get_result_queue() 
# 从task队列取任务,并把结果写入result队列: 
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 queue.Empty: 
 print('task queue is empty.') 
# 处理结束: 
print('worker exit.')

先运行task_master.py,然后再运行task_worker.py

(1)task_master.py运行结果如下

start! 
Put task 7872... 
Put task 6931... 
Put task 1395... 
Put task 8477... 
Put task 8300... 
Put task 1597... 
Put task 8738... 
Put task 8627... 
Put task 1884... 
Put task 2561... 
Try get results... 
Result: 7872 * 7872 = 61968384 
Result: 6931 * 6931 = 48038761 
Result: 1395 * 1395 = 1946025 
Result: 8477 * 8477 = 71859529 
Result: 8300 * 8300 = 68890000 
Result: 1597 * 1597 = 2550409 
Result: 8738 * 8738 = 76352644 
Result: 8627 * 8627 = 74425129 
Result: 1884 * 1884 = 3549456 
Result: 2561 * 2561 = 6558721 
master exit.

(2)task_worker.py运行结果如下

Connect to server 127.0.0.1... 
run task 8640 * 8640... 
run task 7418 * 7418... 
run task 9303 * 9303... 
run task 568 * 568... 
run task 1633 * 1633... 
run task 3583 * 3583... 
run task 3293 * 3293... 
run task 8975 * 8975... 
run task 8189 * 8189... 
run task 731 * 731... 
worker exit.

知识补充

这个简单的Master/Worker模型有什么用"text-align: center">

而Queue之所以能通过网络访问，就是通过QueueManager实现的。由于QueueManager管理的不止一个Queue，所以，要给每个Queue的网络调用接口起个名字，比如get_task_queue。task_worker这里的QueueManager注册的名字必须和task_manager中的一样。对比上面的例子，可以看出Queue对象从另一个进程通过网络传递了过来。只不过这里的传递和网络通信由QueueManager完成。

authkey有什么用"color: #ff0000">总结

以上所述是小编给大家介绍的Python分布式进程中你会遇到的问题解析,希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！

python,分布式进程,python,分布式

广告合作：本站广告合作请联系QQ：858582 申请时备注：广告合作（否则不回）
免责声明：本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除！

评论“Python分布式进程中你会遇到的问题解析”

暂无评论...

www.wwsws.com 伏龙阁资源网

39,976影音资源

44,792技术资源

21,817软件资源

651,128站长资源

Python分布式进程中你会遇到的问题解析

numpy.linspace函数具体使用详解

Django利用cookie保存用户登录信息的简单实现方法

评论“Python分布式进程中你会遇到的问题解析”

稳了！魔兽国服回归的3条重磅消息！官宣时间再确认！

更新日志

友情链接