python manager与basemanager_Python----分布式进程使用(Queue和BaseManager使用)---（搬运）...

骑猪看日落 2022-10-22 11:49 223阅读 0赞

分布式进程

需要模块

multiprocessing和queue模块

使用BaseManager创建分布式管理器

使用Queue创建队列，用于多个进程之间的通信

分布式进程原理

managers子模块支持把多个进程分布到多台机器上

可以写一个服务进程作为调度者，将任务分布到其它多个进程中，然后通过网络通信进行管理

比如爬取图片：一般一个进程负责抓取图片的地址，将地址放在Queue(容器)队列中

另外一个进程负责从Queue队列中取出链接地址进行图片下载和存储到到本地

上述爬取图片的过程就可以做成分布式，一台机器负责获取链接，另外一台机器负责下载存储

上述问题核心：将Queue队列暴露到网络中，让其他机器可以访问

分布式进程实现步骤

建立Queue队列,负责进程之间的通信，任务队列task\_queue,结果队列result\_queue

把第一步中的两个队列在网络中注册,注册时候将队列重新命名

创建一个Queuemanager(BaseManager)的实例manager，相当于一个服务器，给定IP地址、端口和验证码

启动实例manager

访问Queue对象，即创建网络中暴露重命名后的Queue实例

创建任务到本地队列中，自动上传任务到网络队列中，分配给任务进程进行处理

任务进程先从网络中任务队列中取出任务，然后执行，将执行结果放入到网络中的结果队列中

服务进程从结果队列中取出结果，直到执行完所有任务和取出所有的结果，任务进程关闭，然后服务进行关闭

注意

先创建服务进程，再创建任务进程

执行时，先启动服务进程，在创建任务进程，启动任务进程不要超过服务进程取出结果的等待时间

分布式进程实例

创建一个分布式进程，用来完成10次乘法任务

服务进程：

\# 服务进程在windows系统和Linux系统上有所不同

\# 创建一个分布式进程：包括服务进程和任务进程

\# 多个进程之间的通信使用Queue

\# 该代码为服务进程

\# 注意，运行时先运行服务进程，再运行任务进程

\# 任务执行循序：

\# 服务进程和任务进行都创建了相同的两个队列，一个用来放任务，一个用来放结果

\# 第一步：服务进程运行，比如将数字2放进任务队列，任务进程从任务队列中取出数字2

\# 第二步：取出数字，执行任务，就是2\*2=4, 任务执行完后，放进结果队列

\# 第三步：服务进程从结果队列中，取出结果。

\# 第四步：所有任务执行完毕，所有结果都已经取出，最终任务队列和结果队列都是空的了

\# -\*- coding:utf-8 -\*-

import random, queue

from multiprocessing.managers import BaseManager

from multiprocessing import freeze\_support

\# 第一步：定义两个Queue队列，一个用于发送任务，一个接收结果

task\_queue = queue.Queue()

result\_queue = queue.Queue()

\# 创建类似的QueueManager,继承BaseManager,用于后面创建管理器

class QueueManager(BaseManager):

pass

\# 定义两个函数，返回结果就是Queue队列

def return\_task\_queue():

global task\_queue \# 定义成全局变量

return task\_queue \# 返回发送任务的队列

def return\_result\_queue():

global result\_queue

return result\_queue \# 返回接收结果的队列

\# 第二步：把上面创建的两个队列注册在网络上，利用register方法

\# callable参数关联了Queue对象，将Queue对象在网络中暴露

\# 第一个参数是注册在网络上队列的名称

def test():

QueueManager.register('get\_task\_queue', callable=return\_task\_queue)

QueueManager.register('get\_result\_queue', callable=return\_result\_queue)

\# 第三步：绑定端口8001，设置验证口令,这个相当于对象的初始化

\# 绑定端口并填写验证口令，windows下需要填写IP地址，Linux下默认为本地，地址为空

manager = QueueManager(address=('127.0.0.1', 8001), authkey=b'abc') \# 口令必须写成类似b'abc'形式，只写'abc'运行错误

\# 第四步：启动管理器，启动Queue队列，监听信息通道

manager.start()

\# 第五步：通过管理实例的方法获访问网络中的Queue对象

\# 即通过网络访问获取任务队列和结果队列,创建了两个Queue实例，

task = manager.get\_task\_queue()

result = manager.get\_result\_queue()

\# 第六步：添加任务，获取返回的结果

\# 将任务放到Queue队列中

for i in range(10):

n = random.randint(0, 10) \# 返回0到10之间的随机数

print("Put task %s ..." % n)

task.put(n) \# 将n放入到任务队列中

\# 从结果队列中取出结果

print("Try get results...")

for i in range(11): \# 注意，这里结果队列中取结果设置为11次，总共只有10个任务和10个结果，第10次用量确认队列中是不是已经空了

\# 总共循环10次，上面放入了10个数字作为任务

\# 加载一个异常捕获

try:

r = result.get(timeout=5) \# 每次等待5秒，取结果队列中的值

print("Result: %s" % r)

except queue.Empty:

print("result queue is empty.")

\# 最后一定要关闭服务，不然会报管道未关闭的错误

manager.shutdown()

print("master exit.")

if \_\_name\_\_ == '\_\_main\_\_':

\# Windows下多进程可能出现问题，添加以下代码可以缓解

freeze\_support()

print("Start!")

\# 运行服务进程

test()

任务进程

\# coding: utf-8

\# 定义具体的任务进程，具体的工作任务是什么

import time, sys, queue

from multiprocessing.managers import BaseManager

\# 创建类似的QueueManager,继承BaseManager,用于后面创建管理器

class QueueManager(BaseManager):

pass

\# 第一步：使用QueueManager注册用于获取Queue的方法名称

\# 前面服务进程已经将队列名称暴露到网络中，

\# 该任务进程注册时只需要提供名称即可，与服务进程中队列名称一致

QueueManager.register('get\_task\_queue')

QueueManager.register('get\_result\_queue')

\# 第二步：连接到服务器,也就是运行服务进程代码的机器

server\_addr = '127.0.0.1'

print("Connet to server %s..." % server\_addr)

\# 创建一个管理器实例，端口和验证口令保持与服务进程中完全一致

m = QueueManager(address=(server\_addr, 8001), authkey=b'abc')

\# 连接到网络服务器

m.connect()

\# 第三步：从网络上获取Queue对象，并进行本地化，与服务进程是同一个队列

task = m.get\_task\_queue()

result = m.get\_result\_queue()

\# 第四步：从task队列获取任务，并把结果写入到resul队列

for i in range(10):

try:

\# 前面服务进程向task队列中放入了n,这里取出n

\# n和n相乘，并将相乘的算式和结果放入到result队列中去

n = task.get(timeout=1) \# 每次等待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.")

启动分布式进程

先运行服务进程，再运行任务进程

服务进程运行结果

Start!

Put task 2 ...

Put task 0 ...

Put task 6 ...

Put task 8 ...

Put task 4 ...

Put task 1 ...

Try get results...

Result: 2 \* 2 = 4

Result: 0 \* 0 = 0

Result: 6 \* 6 = 36

Result: 8 \* 8 = 64

Result: 4 \* 4 = 16

Result: 1 \* 1 = 1

result queue is empty.

master exit.

Process finished with exit code 0

任务进程运行结果

Connet to server 127.0.0.1...

run task 2 \* 2...

run task 0 \* 0...

run task 6 \* 6...

run task 8 \* 8...

run task 4 \* 4...

run task 1 \* 1...

worker exit.

Process finished with exit code 0

知识补充1

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

知识补充2

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

因为是分布式的环境，放入queue中的数据需要等待Workers机器运算处理后再进行读取，

QueueManager.register(‘get\_task\_queue’, callable=return\_task\_queue)

QueueManager.register(‘get\_result\_queue’, callable=return\_result\_queue)

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

知识补充3

这个简单的Master/Worker模型有什么用？其实这就是一个简单但真正的分布式计算，把代码稍加改造，启动多个worker，就可以把任务分布到几台甚至几十台机器上，比如把计算n\*n的代码换成发送邮件，就实现了邮件队列的异步发送。

Queue对象存储在哪？注意到task\_worker.py中根本没有创建Queue的代码，所以，Queue对象存储在task\_master.py进程中：

![b8159f73ab4f80a95d266340f9f7dc27.png][]

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

authkey有什么用？这是为了保证两台机器正常通信，不被其他机器恶意干扰。如果task\_worker.py的authkey和task\_master.py的authkey不一致，肯定连接不上。

[b8159f73ab4f80a95d266340f9f7dc27.png]: /images/20221022/d4d60e94538b465c80578ebef53c3f37.png