Manager进程之间共享数据-白红宇

本文首发于

本文分为如下部分

引言

初试manager

注意事项

分布式进程

引言

在多进程中，每一个进程都有自己的变量拷贝，所以主进程中的一个变量传入其他进程修改，得到的结果仍然存储于那个进程中，主进程中这个变量其实相当于没有被修改过。为了能让其他进程的修改能够同步到主进程上来，需要创建能在多个进程之间共享的变量。

举一个例子

from multiprocessing import Process def f1(x, l):     x += 1     l.append(2) def f2(x, l):     x -= 2     l.append(3) if __name__ == '__main__':     x = 0     l = [1]     p1 = Process(target=f1, args=(x, l))     p2 = Process(target=f2, args=(x, l))     p1.start()     p2.start()     p1.join()     p2.join()     print(x, l) 复制代码

运行结果为

0 [1] 复制代码

x l都没有被改变，因为它们是放在其他进程中修改的。

初试manager

我们在前面的文章中提到了进程之间共享数据的一些方法，如Queue pipe value，不过multiprocessing模块还提供了一种更加高级的封装，即用Manager来创建变量用于进程之间共享，我们直接来看下面一个例子

from multiprocessing import Process, Manager def f1(ns, l):     ns.x += 1     l.append(2) def f2(ns, l):     ns.x -= 2     l.append(3) if __name__ == '__main__':     manager = Manager()     ns = manager.Namespace()     l = manager.list([1])     ns.x = 0     p1 = Process(target=f1, args=(ns, l))     p2 = Process(target=f2, args=(ns, l))     p1.start()     p2.start()     p1.join()     p2.join()     print(ns, l) 复制代码

结果如下

Namespace(x=-1) [1, 2, 3] 复制代码

上面代码涉及到了manager.Namespace()和manager.list()两个方法，前者可以通过.来创建各种变量，后者专门用来创建list，用manager方法创建出来的变量可以在不同进程之中修改。

manager创造的其他类型详见

注意事项

有时候使用manager仍会发现变量没有被其他进程改变，比如使用manager.Namespace()创建列表修改无效，或manager.list()创建多层列表时里面列表中的元素修改无效。这是因为它们是可变对象，修改时内存地址不变，于是主进程还是读取原来的地址，独到的还是原来的值。

中的代码非常恰当，我就直接贴在这里了

import multiprocessing import time def f(ns, ls, di):     ns.x += 1     ns.y[0] += 1     ns_z = ns.z     ns_z[0] += 1     ns.z = ns_z     ls[0] += 1     ls[1][0] += 1     ls_2 = ls[2]     ls_2[0] += 1     ls[2] = ls_2     di[0] += 1     di[1][0] += 1     di_2 = di[2]     di_2[0] += 1     di[2] = di_2 if __name__ == '__main__':     manager = multiprocessing.Manager()     ns = manager.Namespace()     ns.x = 1     ns.y = [1]     ns.z = [1]     ls = manager.list([1, [1], [1]])     di = manager.dict({
    0: 1, 1: [1], 2:[1]})     print('before', ns, ls, di)     p = multiprocessing.Process(target=f, args=(ns, ls, di))     p.start()     p.join()     print('after', ns, ls, di) 复制代码

运行结果为

before Namespace(x=1, y=[1], z=[1]) [1, [1], [1]] {
    0: 1, 1: [1], 2: [1]} after Namespace(x=2, y=[1], z=[2]) [2, [1], [2]] {
    0: 2, 1: [1], 2: [2]} 复制代码

上面结果ns ls di三部分分开看，每一部分都是第一个第三个变了，第二个没变，读者可以细细感受它们之间的不同

分布式进程

正常分布式应该是在两台电脑上运行的，不过我只有一台电脑，于是就打开两个cmd，运行两个文件，模拟分布式运行了

下面会创建master.py和task1.py两个文件

master.py文件创建一个列表，这个文件负责在列表中append数据。因为只有一个进程于是没有用Process，而是线性执行

task1.py文件需要对master.py文件中的列表进行pop

实现思路如下

master.py文件中列表正常创建并while True循环不断添加元素

master.py文件要设置一个账号密码，将这个list变量暴露出来

task1.py文件要通过账号密码连接到master.py文件，并提取list变量

master.py文件内容如下

import random, time from multiprocessing.managers import BaseManager as bm l = [] def return_l(): return l if __name__ == '__main__': # 三步固定这么做即可     bm.register('get_l', callable = return_l)     m = bm(address = ('127.0.0.1', 5000), authkey = b'abc')     m.start()     new_l = m.get_l() while True:         new = random.randint(0, 100)         new_l.append(new)         print('produce {} now all {}'.format(new, new_l))         time.sleep(2 * random.random())     m.shutdown()     print('master exit') 复制代码

task1.py文件内容如下

import random import time from multiprocessing.managers import BaseManager as bm if __name__ == '__main__':     bm.register('get_l')     m = bm(address = ('127.0.0.1', 5000), authkey = b'abc')     m.connect()     l = m.get_l() while True:         print('drop {}'.format(l.pop()))         time.sleep(3 * random.random()) 复制代码

其实看着上面代码结合之前提供的思路应该就能知道分布式是怎么做的了，下面讲一下运行如何操作。

保存好这两个文件后，分别在这两个文件所在位置打开cmd，下面我就分别称为cmdm和cmdt了

在cmdm中输入

python master.py 复制代码

这时就会发现程序开始运行，list中元素也越来越多。然后在cmdt中输入

python task1.py 复制代码

就会发现cmdm中的list元素会开始减少。

这个过程其实就是运行两个文件，只要保证master.py运行在前即可。运行两个文件其实相当于开启两个进程，在两台电脑中开启也是一样。

注意：上面代码在windows系统中运行。linux下分布式进程可以参考，这篇文章中的代码在windows下是无法运行的，将其改写成windows下的版本参考

欢迎关注我的知乎专栏

专栏主页：

专栏目录：

版本说明：