A-A+

python 进程间共享数据 multiprocessing 通信问题 — Manager

2019年01月10日 18:52 汪洋大海 暂无评论 共6187字 (阅读5,519 views次)

Python中进程间共享数据，处理基本的queue，pipe和value+array外，还提供了更高层次的封装。使用multiprocessing.Manager可以简单地使用这些高级接口。
Manager()返回的manager对象控制了一个server进程，此进程包含的python对象可以被其他的进程通过proxies来访问。从而达到多进程间数据通信且安全。
Manager支持的类型有list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。

Manager的dict，list使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import multiprocessing
import time
def worker(d, key, value):
    d[key] = value
 
if __name__ == '__main__':
    mgr = multiprocessing.Manager()
    d = mgr.dict()
    jobs = [ multiprocessing.Process(target=worker, args=(d, i, i*2))
             for i in range(10)
             ]
    for j in jobs:
        j.start()
    for j in jobs:
        j.join()
    print ('Results:' )
    for key, value in enumerate(dict(d)):
        print("%s=%s" % (key, value))
 
# the output is :
# Results:
# 0=0
# 1=1
# 2=2
# 3=3
# 4=4
# 5=5
# 6=6
# 7=7
# 8=8
# 9=9

文章来源：https://www.cnblogs.com/caodneg7/p/9520069.html

Python 多进程默认不能共享全局变量
主进程与子进程是并发执行的，进程之间默认是不能共享全局变量的(子进程不能改变主进程中全局变量的值)。如果要共享全局变量需要用（multiprocessing.Value("d",10.0)，数值）（multiprocessing.Array("i",[1,2,3,4,5])，数组）（multiprocessing.Manager().dict()，字典）（multiprocessing.Manager().list(range(5))）。进程通信（进程之间传递数据）用进程队列（multiprocessing.Queue()，单向通信），管道（ multiprocessing.Pipe() ，双向通信）。

这里要注意多线程之间数据共享、多进程之间数据共享。。。。。。。多线程用全局变量(global)
全局变量，主进程与子进程是并发执行的，他们不能共享全局变量(子进程不能改变主进程中全局变量的值)

由于进程之间不共享内存，所以进程之间的通信不能像线程之间直接引用，因而需要采取一些策略来完成进程之间的数据通信。
本文记录使用 Manager 来完成进程间通信的方式。
首先描述需求：
    场景：顶层逻辑负责管理，我们定义为C，由C启动A、B两个进程联合完成功能
    需求：A、B联合工作过程中的数据通信

python 多进程

解决思路：

利用顶层C创建一个 Manager，由 Manager 提供数据池分发给A、B使用，从而完成两个进程之间的通信。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

#-*-encoding:utf-8-*-
from multiprocessing import Process, Manager
from time import sleep
 
 
def thread_a_main(sync_data_pool):  # A 进程主函数，存入100+的数
    for ix in range(100, 105):
        sleep(1)
        sync_data_pool.append(ix)
 
 
def thread_b_main(sync_data_pool):  # B 进程主函数，存入300+的数
    for ix in range(300, 309):
        sleep(0.6)
        sync_data_pool.append(ix)
 
 
def _test_case_000():  # 测试用例
    manager = Manager()  # multiprocessing 中的 Manager 是一个工厂方法，直接获取一个 SyncManager 的实例
    sync_data_pool = manager.list()  # 利用 SyncManager 的实例来创建同步数据池
    Process(target=thread_a_main, args=(sync_data_pool, )).start()  # 创建并启动 A 进程
    Process(target=thread_b_main, args=(sync_data_pool, )).start()  # 创建并启动 B 进程
    for ix in range(6):  # C 进程（主进程）中实时的去查看数据池中的数据
        sleep(1)
        print(sync_data_pool)
 
 
if '__main__' == __name__:  # 养成好习惯，将测试用例单独列出
    _test_case_000()

输出结果：

[]
[300, 100, 301]
[300, 100, 301, 101, 302, 303]
[300, 100, 301, 101, 302, 303, 102, 304]
[300, 100, 301, 101, 302, 303, 102, 304, 305, 103, 306]
[300, 100, 301, 101, 302, 303, 102, 304, 305, 103, 306, 104, 307, 308]
从结果来看， 我们的300+和100+的数据并行的被插入到同一数据池中了，也就是说，通信的目的达到了

只是从目前的方式来看，接收方必须主动的去查询数据池，类似于信箱的方式了
除了注释中的内容外，还需要注意：
    1、manager 要在顶层创建
    2、同步数据池是通过Manager自身的工厂方法创建的，这里 manager.list() 调用一次即产生一个新的数据池，而不是返回同一个数据池实例，所以数据池的实例需要做好管理
    3、可以用的数据格式不仅list，可以选择如下（参考multiprocessing.managers）

class SyncManager(BaseManager):
    def BoundedSemaphore(self, value: Any = ...) -> threading.BoundedSemaphore: ...
    def Condition(self, lock: Any = ...) -> threading.Condition: ...
    def Event(self) -> threading.Event: ...
    def Lock(self) -> threading.Lock: ...
    def Namespace(self) -> _Namespace: ...
    def Queue(self, maxsize: int = ...) -> queue.Queue: ...
    def RLock(self) -> threading.RLock: ...
    def Semaphore(self, value: Any = ...) -> threading.Semaphore: ...
    def Array(self, typecode: Any, sequence: Sequence[_T]) -> Sequence[_T]: ...
    def Value(self, typecode: Any, value: _T) -> _T: ...
    def dict(self, sequence: Mapping[_KT, _VT] = ...) -> Dict[_KT, _VT]: ...
    def list(self, sequence: Sequence[_T] = ...) -> List[_T]: ...

进程之间共享数据(数值型)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import multiprocessing
 
def  func(num):
    num.value=10.78  #子进程改变数值的值，主进程跟着改变
 
if  __name__=="__main__":
    num=multiprocessing.Value("d",10.0) # d表示数值,主进程与子进程共享这个value。（主进程与子进程都是用的同一个value）
    print(num.value)
 
    p=multiprocessing.Process(target=func,args=(num,))
    p.start()
    p.join()
 
    print(num.value)

进程之间共享数据(数组型)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import multiprocessing
 
def  func(num):
    num[2]=9999   #子进程改变数组，主进程跟着改变
 
if  __name__=="__main__":
    num=multiprocessing.Array("i",[1,2,3,4,5])   #主进程与子进程共享这个数组
    print(num[:])
 
    p=multiprocessing.Process(target=func,args=(num,))
    p.start() 
    p.join()
 
    print(num[:])

进程之间共享数据(dict,list)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import multiprocessing
 
def func(mydict,mylist):
    mydict["index1"]="aaaaaa"   #子进程改变dict,主进程跟着改变
    mydict["index2"]="bbbbbb"
    mylist.append(11)        #子进程改变List,主进程跟着改变
    mylist.append(22)
    mylist.append(33)
 
if __name__=="__main__":
    with multiprocessing.Manager() as MG:   #重命名
        mydict=multiprocessing.Manager().dict()   #主进程与子进程共享这个字典
        mylist=multiprocessing.Manager().list(range(5))   #主进程与子进程共享这个List
 
        p=multiprocessing.Process(target=func,args=(mydict,mylist))
        p.start()
        p.join()
 
        print(mylist)
        print(mydict)

-----------------------------------------------------------------
Value、Array是通过共享内存的方式共享数据
Manager是通过共享进程的方式共享数据。

Value\Array
实例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import multiprocessing
#Value/Array
def func1(a,arr):
    a.value=3.14
    for i in range(len(arr)):
        arr[i]=-arr[i]
if __name__ == '__main__':
    num=multiprocessing.Value('d',1.0)#num=0
    arr=multiprocessing.Array('i',range(10))#arr=range(10)
    p=multiprocessing.Process(target=func1,args=(num,arr))
    p.start()
    p.join()
    print num.value
    print arr[:]

执行结果：

1
2
3
4
5

3.14
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]
1
2
3

Manager管理的共享数据类型有：Value、Array、dict、list、Lock、Semaphore等等，同时Manager还可以共享类的实例对象。
实例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from multiprocessing import Process,Manager
def func1(shareList,shareValue,shareDict,lock):
    with lock:
        shareValue.value+=1
        shareDict[1]='1'
        shareDict[2]='2'
        for i in xrange(len(shareList)):
            shareList[i]+=1
 
if __name__ == '__main__':
    manager=Manager()
    list1=manager.list([1,2,3,4,5])
    dict1=manager.dict()
    array1=manager.Array('i',range(10))
    value1=manager.Value('i',1)
    lock=manager.Lock()
    proc=[Process(target=func1,args=(list1,value1,dict1,lock)) for i in xrange(20)]
    for p in proc:
        p.start()
    for p in proc:
        p.join()
    print list1
    print dict1
    print array1
    print value1

执行结果：

1
2
3
4

[21, 22, 23, 24, 25]
{1: '1', 2: '2'}
array('i', [0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
Value('i', 21)

通过Manager进程间共享实例对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

from multiprocessing import Process,Value,Lock
from multiprocessing.managers import BaseManager
class Employee(object):
    def __init__(self,name,salary):
        self.name=name
        self.salary=Value('i',salary)
    def increase(self):
        self.salary.value+=100
    def getPay(self):
        return self.name+':'+str(self.salary.value)
class MyManager(BaseManager):
    pass
def Manager2():
    m=MyManager()
    m.start()
    return m
MyManager.register('Employee',Employee)
 
def func1(em,lock):
    with lock:
        em.increase()
 
if __name__ == '__main__':
    manager=Manager2()
    em=manager.Employee('zhangsan',1000)
    lock=Lock()
    proces=[Process(target=func1,args=(em,lock))for i in xrange(10)]
    for p in proces:
        p.start()
    for p in proces:
        p.join()
    print em.getPay()

资料来源：https://blog.csdn.net/houyanhua1/article/details/78236514
https://blog.csdn.net/houyanhua1/article/details/78244288
https://blog.csdn.net/lechunluo3/article/details/79005910

布施恩德可便相知重

微信扫一扫打赏

支付宝扫一扫打赏

×
标签：

给我留言